diff options
| -rw-r--r-- | python/fatcat_tools/workers/changelog.py | 5 | ||||
| -rw-r--r-- | python/fatcat_tools/workers/worker_common.py | 2 |
2 files changed, 5 insertions, 2 deletions
diff --git a/python/fatcat_tools/workers/changelog.py b/python/fatcat_tools/workers/changelog.py index ccfbcf19..d07b5988 100644 --- a/python/fatcat_tools/workers/changelog.py +++ b/python/fatcat_tools/workers/changelog.py @@ -1,7 +1,6 @@ import json import time -from itertools import islice from fatcat_tools.workers.worker_common import FatcatWorker, most_recent_message from pykafka.common import OffsetType @@ -33,7 +32,7 @@ class FatcatChangelogWorker(FatcatWorker): else: self.offset = 1 - with topic.get_sync_producer() as producer: + with topic.get_producer() as producer: while True: latest = int(self.api.get_changelog(limit=1)[0].index) if latest > self.offset: @@ -83,6 +82,8 @@ class FatcatEntityUpdatesWorker(FatcatWorker): compacted_topic=True, ) + # using a sync producer to try and avoid racey loss of delivery (aka, + # if consumer group updated but produce didn't stick) with release_topic.get_sync_producer() as producer: for msg in consumer: cle = json.loads(msg.value.decode('utf-8')) diff --git a/python/fatcat_tools/workers/worker_common.py b/python/fatcat_tools/workers/worker_common.py index 66150644..daae91b3 100644 --- a/python/fatcat_tools/workers/worker_common.py +++ b/python/fatcat_tools/workers/worker_common.py @@ -5,7 +5,9 @@ import csv import json import itertools import fatcat_client +from itertools import islice from pykafka import KafkaClient +from pykafka.common import OffsetType from fatcat_client.rest import ApiException |