手动提交偏移量到kafka主题的正确方法是什么

Question

我有一个消费者脚本，它处理每条消息并手动向主题提交偏移量。

CONSUMER = KafkaConsumer(
    KAFKA_TOPIC,
    bootstrap_servers=[KAFKA_SERVER],
    auto_offset_reset="earliest",
    max_poll_records=100,
    enable_auto_commit=False,
    group_id=CONSUMER_GROUP,
    # Use the RoundRobinPartition method
    partition_assignment_strategy=[RoundRobinPartitionAssignor],
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

while True:
    count += 1
    LOGGER.info("--------------Poll {0}---------".format(count))
    for msg in CONSUMER:
        # Process msg.value
        # Commit offset to topic
        tp = TopicPartition(msg.topic, msg.partition)
        offsets = {tp: OffsetAndMetadata(msg.offset, None)}
        CONSUMER.commit(offsets=offsets)

处理每条消息所需的时间

我收到此错误错误：

kafka.errors.CommitFailedError: CommitFailedError: Commit cannot be completed since the group has already
            rebalanced and assigned the partitions to another member.
            This means that the time between subsequent calls to poll()
            was longer than the configured max_poll_interval_ms, which
            typically implies that the poll loop is spending too much
            time message processing. You can address this either by
            increasing the rebalance timeout with max_poll_interval_ms,
            or by reducing the maximum size of batches returned in poll()
            with max_poll_records.


Process finished with exit code 1

期望：

a) 如何解决这个错误？

b) 如何确保我的手动提交工作正常？

c) 提交offset的正确方式。

我已经完成了这个，但Difference between session.timeout.ms and max.poll.interval.ms for Kafka 0.10.0.0 and later versions 了解我的问题，非常感谢任何关于调整投票、会话或心跳时间的帮助。

Apache 卡夫卡：2.11-2.1.0 卡夫卡蟒：1.4.4

Answer 1

消费者的session.timeout.ms 应该小于Kafka 代理上的group.max.session.timeout.ms。