0

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?
4

0 に答える 0