From 83ae1d85d093f1f37bba74283b1d689bb1f1c346 Mon Sep 17 00:00:00 2001
From: Bryan Newbold <bnewbold@robocracy.org>
Date: Mon, 12 Apr 2021 15:27:34 -0700
Subject: es worker: ensure kafka messages get cleared

---
 python/fatcat_tools/workers/elasticsearch.py | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'python')

diff --git a/python/fatcat_tools/workers/elasticsearch.py b/python/fatcat_tools/workers/elasticsearch.py
index d9edb276..b8735a37 100644
--- a/python/fatcat_tools/workers/elasticsearch.py
+++ b/python/fatcat_tools/workers/elasticsearch.py
@@ -141,6 +141,8 @@ class ElasticsearchReleaseWorker(FatcatWorker):
 
             # if only WIP entities, then skip
             if not bulk_actions:
+                for msg in batch:
+                    consumer.store_offsets(message=msg)
                 continue
 
             print("Upserting, eg, {} (of {} {} in elasticsearch)".format(key, len(batch), self.entity_type.__name__), file=sys.stderr)
-- 
cgit v1.2.3