fix worker code

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