From a7736d91665f6a98090cd448d02f1542aec6c180 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 10 Dec 2019 10:21:21 -0800 Subject: factor out some basic kafka helpers --- python/fatcat_tools/kafka.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 python/fatcat_tools/kafka.py (limited to 'python/fatcat_tools/kafka.py') diff --git a/python/fatcat_tools/kafka.py b/python/fatcat_tools/kafka.py new file mode 100644 index 00000000..53b62a37 --- /dev/null +++ b/python/fatcat_tools/kafka.py @@ -0,0 +1,22 @@ + +from confluent_kafka import Consumer, Producer, KafkaException + + +def kafka_fail_fast(err, msg): + if err is not None: + print("Kafka producer delivery error: {}".format(err)) + print("Bailing out...") + # TODO: should it be sys.exit(-1)? + raise KafkaException(err) + +def simple_kafka_producer(kafka_hosts): + + kafka_config = { + 'bootstrap.servers': kafka_hosts, + 'message.max.bytes': 20000000, # ~20 MBytes; broker-side max is ~50 MBytes + 'delivery.report.only.error': True, + 'default.topic.config': { + 'request.required.acks': -1, + }, + } + return Producer(kafka_config) -- cgit v1.2.3