consumer.commit()
auto.commit=True モードで実行されているコンシューマーの各メッセージでメソッドを呼び出しています。ローカルで試したところ、データの損失やデータの重複なしに機能しました。
commit()
この方法は消費者にどのような影響を与えることができますか?
class KafkaConsumer(object):
def __init__(self, topics: list[str], **kwargs: Any):
config = {
**kwargs,
}
self.consumer = DeserializingConsumer(config)
self.consumer.subscribe(topics=topics)
def consume(self, poll_timeout_secs: float = 1.0):
while True:
msg = self.consumer.poll(timeout=poll_timeout_secs)
if msg is None:
continue
if msg.error():
raise KafkaException(msg.error())
else:
yield msg
kafka_consumer = KafkaConsumer(topics=topics, **kwargs) # passed "enable.auto.commit":True
for message in kafka_consumer.consume():
event = message.value()
logger.info(f"message {event}")
kafka_consumer.commit() # what will be the effect of calling this?