diff options
| author | Bryan Newbold <bnewbold@robocracy.org> | 2019-11-13 00:28:13 -0800 |
|---|---|---|
| committer | Bryan Newbold <bnewbold@robocracy.org> | 2019-11-15 16:46:26 -0800 |
| commit | 31f237a6150f22676e93902a3597461ea954dc2c (patch) | |
| tree | f2f688ad2a49c0edbee4fe2eff4ee096a368b780 /python/fatcat_tools | |
| parent | a4db9ee2e18a18b23eb7ece484f95914421f877d (diff) | |
| download | fatcat-31f237a6150f22676e93902a3597461ea954dc2c.tar.gz fatcat-31f237a6150f22676e93902a3597461ea954dc2c.zip | |
crude support for 'sandcrawler' kafka topics
Diffstat (limited to 'python/fatcat_tools')
| -rw-r--r-- | python/fatcat_tools/importers/common.py | 5 |
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: |