I'm using kafka-python==2.0.2
, and have disabled auto_commit
but still if I don't commit through code, offsets are automatically getting committed
In the below code even if I comment out self.consumer.commit_async(callback= ....
, offsets are still getting committed
class KafkaMessageConsumer:
def __init__(self, bootstrap_servers: str, topic: str, group_id: str, offset_reset_strategy: str):
self.bootstrap_servers: str = bootstrap_servers
self.topic: str = topic
self.group_id: str = group_id
self.consumer: KafkaConsumer = KafkaConsumer(topic, bootstrap_servers=bootstrap_servers, group_id=group_id,
enable_auto_commit=False, auto_offset_reset=offset_reset_strategy)
def consume_messages(self, consumer_poll_timeout: int, max_poll_records: int,
message_handler: MessageHandlerImpl = MessageHandlerImpl()):
try:
while True:
try:
msg_pack = self.consumer.poll(timeout_ms=consumer_poll_timeout, max_records=max_poll_records)
if bool(msg_pack):
for topic_partition, messages in msg_pack.items():
message_handler.process_messages(messages)
self.consumer.commit_async(callback=(lambda offsets, response: log.error(
f"Error while committing offset in async due to: {response}", exc_info=True) if isinstance(
response, Exception) else log.debug(f"Successfully committed offsets: {offsets}")))
except Exception as e:
log.error(f"Error while consuming/processing message due to: {e}", exc_info=True)
finally:
log.error("Something went wrong, closing consumer...........")
self.consumer.close()
Is this a proper way to disable auto commit and commit manually?
poll()
always moves forward, regardless of committing. It just does not write offsets back to Kafka.
If you restart your app, then would it pickup the same data again and again.