aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--python/fatcat_tools/workers/changelog.py5
-rw-r--r--python/fatcat_tools/workers/worker_common.py2
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