Kafka消费者多次收到相同的消息答案

【问题标题】：Kafka consumer receiving same message multiple timesKafka消费者多次收到相同的消息
【发布时间】：2017-02-05 12:17:32
【问题描述】：

我最近开始使用 kafka 来阅读通过网络爬虫传来的文档。我注意到的是，当我处理几百万个文档时，消费者会一遍又一遍地处理相同的消息。看起来数据由于某种原因没有被提交。当我用几百条消息测试消费者时，情况并非如此。

我在 java 中使用 kafka 高级消费者客户端代码。我正在使用在与分区数相等的线程数上运行的消费者组。因此，每个线程都被分配到一个分区。这是用于轮询数据的代码 sn-p。

while (true) {
    try{
        if(consumerDao.canPollTopic()){
            ConsumerRecords records = 
              consumer.poll(this.config.getPropertyAsIneger(IPreProcessorConstant.KAFKA_POLL_COUNT));
            for (ConsumerRecord record : records) {
                if(record.value()!=null){
                    TextAnalysisRequest textAnalysisObj = record.value();
                    if(textAnalysisObj!=null){
                        PostProcessRequest req = new PostProcessRequest();
                        req.setRequest(this.getRequest(textAnalysisObj));
                        PreProcessorUtil.submitPostProcessRequest(req, config);
                    }
                }
            }
        }else{
            Thread.sleep(this.config.getPropertyAsIneger(IPreProcessorConstant.KAFKA_POLL_SLEEP));
        }
    }catch(Exception ex){
        LOGGER.error("Error in Full Consumer group worker", ex);

    }
}

这是我正在设置的 kafka 消费者配置参数。其余为默认值。


consumer.auto.commit=true
consumer.auto.commit.interval=1000
consumer.session.timeout=180000
consumer.poll.records=2147483647
consumer.request.timeout=181000

这是完整的消费者配置：


metric.reporters = 
metadata.max.age.ms = 300000
partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
reconnect.backoff.ms = 50
sasl.kerberos.ticket.renew.window.factor = 0.8
max.partition.fetch.bytes = 1048576
bootstrap.servers = [kafkahost1:9092, kafkahost2:9092]
ssl.keystore.type = JKS
enable.auto.commit = true
sasl.mechanism = GSSAPI
interceptor.classes = null
exclude.internal.topics = true
ssl.truststore.password = null
client.id =ssl.endpoint.identification.algorithm = null
max.poll.records = 2147483647
check.crcs = true
request.timeout.ms = 181000
heartbeat.interval.ms = 3000
auto.commit.interval.ms = 1000
receive.buffer.bytes = 65536
ssl.truststore.type = JKS
ssl.truststore.location = null
ssl.keystore.password = null
fetch.min.bytes = 1
send.buffer.bytes = 131072
value.deserializer = class com.test.preprocessor.consumer.serializer.KryoObjectSerializer
group.id = full_group
retry.backoff.ms = 100
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
ssl.trustmanager.algorithm = PKIX
ssl.key.password = null
fetch.max.wait.ms = 500
sasl.kerberos.min.time.before.relogin = 60000
connections.max.idle.ms = 540000
session.timeout.ms = 180000
metrics.num.samples = 2
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
ssl.protocol = TLS
ssl.provider = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.keystore.location = null
ssl.cipher.suites = null
security.protocol = PLAINTEXT
ssl.keymanager.algorithm = SunX509
metrics.sample.window.ms = 30000
auto.offset.reset = latest

我的示例 kafka 队列有 8 个分区和 2 个复制因子。

server.properties 中的日志保留期设置为 168 小时。

log.retention.hours=168
log.roll.hours=168

不知道我在这里缺少什么。

【问题讨论】：

也不确定。但是，只要您的消费者正在运行，他们就会在内存中保持偏移量——提交偏移量仅用于在消费者关闭或崩溃时保留偏移量。因此，在每个poll() 之后，偏移量都会增加。
@MatthiasJ.Sax ...谢谢。根据日志，我可以看到消费者组下的不同线程正在接收相同的消息。是否有可能将相同的消息传递到多个分区？我可以看到偏移量正在增加。我的目的是尝试了解这背后的可能原因，以便我可以避免。配置参数中是否有任何调整可以避免这种情况？感谢您的反馈。
如果所有线程都使用相同的group.id，则每个分区应该分配给一个线程。只有在重新平衡的情况下才会出现重复消息（如果新消费者加入或离开组）。你能仔细检查一下分区分配吗？
@MatthiasJ.Sax .. 我增加了分区/消费者线程的数量，并在队列中添加了几百万条消息。仔细检查日志后，我发现以下警告，它解释了消费者循环问题。“组 full_group 的自动偏移提交失败：提交无法完成，因为该组已经重新平衡并将分区分配给另一个成员”。我ve max.poll.records=20000 和 session.timeout.ms=180000ms。我想我需要解决这个问题以获得最佳设置。只是想知道是否有围绕这个的最佳实践/拇指规则？
如果您以 1:1 线程与分区比率的同步方式发送/接收，我认为将 maxInFlightConnections 值设置为 1 可能会有所帮助。默认值为 5。

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

我猜是分区分配的问题。

【讨论】：

【解决方案2】：

我将消费者属性中的auto.commit.interval.ms=8000 从3000 增加到8000。这解决了重复记录问题。

【讨论】：