summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorBryan Newbold <bnewbold@robocracy.org>2019-11-13 00:28:13 -0800
committerBryan Newbold <bnewbold@robocracy.org>2019-11-15 16:46:26 -0800
commit31f237a6150f22676e93902a3597461ea954dc2c (patch)
treef2f688ad2a49c0edbee4fe2eff4ee096a368b780
parenta4db9ee2e18a18b23eb7ece484f95914421f877d (diff)
downloadfatcat-31f237a6150f22676e93902a3597461ea954dc2c.tar.gz
fatcat-31f237a6150f22676e93902a3597461ea954dc2c.zip
crude support for 'sandcrawler' kafka topics
-rw-r--r--python/fatcat_tools/importers/common.py5
1 files changed, 3 insertions, 2 deletions
diff --git a/python/fatcat_tools/importers/common.py b/python/fatcat_tools/importers/common.py
index 74595790..c0740f94 100644
--- a/python/fatcat_tools/importers/common.py
+++ b/python/fatcat_tools/importers/common.py
@@ -711,6 +711,7 @@ class KafkaJsonPusher(RecordPusher):
kafka_env,
topic_suffix,
group,
+ kafka_namespace=kwargs.get('kafka_namespace', 'fatcat')
)
self.poll_interval = kwargs.get('poll_interval', 5.0)
self.consume_batch_size = kwargs.get('consume_batch_size', 100)
@@ -758,8 +759,8 @@ class KafkaJsonPusher(RecordPusher):
return counts
-def make_kafka_consumer(hosts, env, topic_suffix, group):
- topic_name = "fatcat-{}.{}".format(env, topic_suffix)
+def make_kafka_consumer(hosts, env, topic_suffix, group, kafka_namespace="fatcat"):
+ topic_name = "{}-{}.{}".format(kafka_namespace, env, topic_suffix)
def fail_fast(err, partitions):
if err is not None: