From e456e6a17eb9655afc4dc8146f50a7dba4fd8601 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Wed, 13 Nov 2019 16:42:05 -0800 Subject: refactor consume_topic name out of make_kafka_consumer() Best to do this in wrapping code for full flexibility. --- python/sandcrawler/workers.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'python/sandcrawler/workers.py') diff --git a/python/sandcrawler/workers.py b/python/sandcrawler/workers.py index 3b46cb7..c1203e1 100644 --- a/python/sandcrawler/workers.py +++ b/python/sandcrawler/workers.py @@ -310,15 +310,15 @@ class ZipfilePusher(RecordPusher): class KafkaJsonPusher(RecordPusher): - def __init__(self, worker, kafka_hosts, kafka_env, topic_suffix, group, **kwargs): + def __init__(self, worker, kafka_hosts, consume_topic, group, **kwargs): self.counts = Counter() self.worker = worker self.consumer = make_kafka_consumer( kafka_hosts, - kafka_env, - topic_suffix, + consume_topic, group, ) + self.push_batches = kwargs.get('push_batches', False) self.poll_interval = kwargs.get('poll_interval', 5.0) self.batch_size = kwargs.get('batch_size', 100) if self.batch_size in (0, 1): @@ -375,8 +375,8 @@ class KafkaJsonPusher(RecordPusher): return self.counts -def make_kafka_consumer(hosts, env, topic_suffix, group): - topic_name = "fatcat-{}.{}".format(env, topic_suffix) +def make_kafka_consumer(hosts, consume_topic, group): + topic_name = consume_topic def fail_fast(err, partitions): if err is not None: -- cgit v1.2.3