summaryrefslogtreecommitdiffstats
path: root/python/fatcat_tools/importers/common.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/fatcat_tools/importers/common.py')
-rw-r--r--python/fatcat_tools/importers/common.py21
1 files changed, 17 insertions, 4 deletions
diff --git a/python/fatcat_tools/importers/common.py b/python/fatcat_tools/importers/common.py
index 6815a155..e936477c 100644
--- a/python/fatcat_tools/importers/common.py
+++ b/python/fatcat_tools/importers/common.py
@@ -732,8 +732,8 @@ class KafkaBs4XmlPusher(RecordPusher):
batch = self.consumer.consume(
num_messages=self.consume_batch_size,
timeout=self.poll_interval)
- print("... got {} kafka messages ({}sec poll interval)".format(
- len(batch), self.poll_interval))
+ print("... got {} kafka messages ({}sec poll interval) {}".format(
+ len(batch), self.poll_interval, self.importer.counts))
if not batch:
if datetime.datetime.now() - last_push > datetime.timedelta(minutes=5):
# it has been some time, so flush any current editgroup
@@ -779,10 +779,12 @@ class KafkaJsonPusher(RecordPusher):
)
self.poll_interval = kwargs.get('poll_interval', 5.0)
self.consume_batch_size = kwargs.get('consume_batch_size', 100)
+ self.force_flush = kwargs.get('force_flush', False)
def run(self):
count = 0
last_push = datetime.datetime.now()
+ last_force_flush = datetime.datetime.now()
while True:
# Note: this is batch-oriented, because underlying importer is
# often batch-oriented, but this doesn't confirm that entire batch
@@ -796,13 +798,24 @@ class KafkaJsonPusher(RecordPusher):
batch = self.consumer.consume(
num_messages=self.consume_batch_size,
timeout=self.poll_interval)
- print("... got {} kafka messages ({}sec poll interval)".format(
- len(batch), self.poll_interval))
+ print("... got {} kafka messages ({}sec poll interval) {}".format(
+ len(batch), self.poll_interval, self.importer.counts))
+ if self.force_flush:
+ # this flushing happens even if there have been 'push' events
+ # more recently. it is intended for, eg, importers off the
+ # ingest file stream *other than* the usual file importer. the
+ # web/HTML and savepapernow importers get frequent messages,
+ # but rare 'want() == True', so need an extra flush
+ if datetime.datetime.now() - last_force_flush > datetime.timedelta(minutes=5):
+ self.importer.finish()
+ last_push = datetime.datetime.now()
+ last_force_flush = datetime.datetime.now()
if not batch:
if datetime.datetime.now() - last_push > datetime.timedelta(minutes=5):
# it has been some time, so flush any current editgroup
self.importer.finish()
last_push = datetime.datetime.now()
+ last_force_flush = datetime.datetime.now()
#print("Flushed any partial import batch: {}".format(self.importer.counts))
continue
# first check errors on entire batch...