From b72c18e3518e827bd09044deaadcbf0b0ca50335 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Fri, 1 Oct 2021 17:39:40 -0700 Subject: kafka import: optional 'force-flush' mode for some importers Behavior and motivation described in the kafka json import comment. --- python/fatcat_import.py | 3 +++ 1 file changed, 3 insertions(+) (limited to 'python/fatcat_import.py') diff --git a/python/fatcat_import.py b/python/fatcat_import.py index 7e790fa4..b82e81c7 100755 --- a/python/fatcat_import.py +++ b/python/fatcat_import.py @@ -159,6 +159,7 @@ def run_ingest_web(args): "fatcat-{}-ingest-web-result".format(args.kafka_env), kafka_namespace="sandcrawler", consume_batch_size=args.batch_size, + force_flush=True, ).run() else: JsonLinePusher(iwri, args.json_file).run() @@ -176,6 +177,7 @@ def run_savepapernow_file(args): "fatcat-{}-savepapernow-file-result".format(args.kafka_env), kafka_namespace="sandcrawler", consume_batch_size=args.batch_size, + force_flush=True, ).run() else: JsonLinePusher(ifri, args.json_file).run() @@ -193,6 +195,7 @@ def run_savepapernow_web(args): "fatcat-{}-savepapernow-web-result".format(args.kafka_env), kafka_namespace="sandcrawler", consume_batch_size=args.batch_size, + force_flush=True, ).run() else: JsonLinePusher(ifri, args.json_file).run() -- cgit v1.2.3