summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorDana Powers <dana.powers@gmail.com>2017-03-06 16:09:22 -0800
committerGitHub <noreply@github.com>2017-03-06 16:09:22 -0800
commit77e1ba36b330268c2db10e863da44484988d781c (patch)
tree8e8b03645188c2424731f3cb22d0c4ab666a9aaf
parent634d24fa5bf1c258acac7956f301b0176c325f89 (diff)
downloadkafka-python-77e1ba36b330268c2db10e863da44484988d781c.tar.gz
Disable default consumer group (#1016)
-rw-r--r--kafka/consumer/group.py4
-rw-r--r--test/test_consumer_group.py2
2 files changed, 4 insertions, 2 deletions
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 1addcc2..344e7e3 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -53,7 +53,7 @@ class KafkaConsumer(six.Iterator):
partition assignment (if enabled), and to use for fetching and
committing offsets. If None, auto-partition assignment (via
group coordinator) and offset commits are disabled.
- Default: 'kafka-python-default-group'
+ Default: None
key_deserializer (callable): Any callable that takes a
raw message key and returns a deserialized key.
value_deserializer (callable): Any callable that takes a
@@ -215,7 +215,7 @@ class KafkaConsumer(six.Iterator):
DEFAULT_CONFIG = {
'bootstrap_servers': 'localhost',
'client_id': 'kafka-python-' + __version__,
- 'group_id': 'kafka-python-default-group',
+ 'group_id': None,
'key_deserializer': None,
'value_deserializer': None,
'fetch_max_wait_ms': 500,
diff --git a/test/test_consumer_group.py b/test/test_consumer_group.py
index 9d9be60..885ae83 100644
--- a/test/test_consumer_group.py
+++ b/test/test_consumer_group.py
@@ -54,12 +54,14 @@ def test_group(kafka_broker, topic):
stop = {}
threads = {}
messages = collections.defaultdict(list)
+ group_id = 'test-group-' + random_string(6)
def consumer_thread(i):
assert i not in consumers
assert i not in stop
stop[i] = threading.Event()
consumers[i] = KafkaConsumer(topic,
bootstrap_servers=connect_str,
+ group_id=group_id,
heartbeat_interval_ms=500)
while not stop[i].is_set():
for tp, records in six.itervalues(consumers[i].poll(100)):