refactor consume_topic name out of make_kafka_consumer()

Best to do this in wrapping code for full flexibility.

1 files changed, 5 insertions, 5 deletions

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: