From ef50d665b31a8cb1b0ddc38671b3334b0fa6fc14 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Wed, 14 Nov 2018 23:23:07 -0800 Subject: fix worker code --- python/fatcat_tools/workers/changelog.py | 5 +++-- python/fatcat_tools/workers/worker_common.py | 2 ++ 2 files changed, 5 insertions(+), 2 deletions(-) (limited to 'python') 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 -- cgit v1.2.3