java - Kafka消费者多次收到同一条消息

Question

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

我在 java 中使用 kafka 高级消费者客户端代码。我正在使用在等于分区数的线程数上运行的消费者组。所以每个线程都被分配到一个分区。这是轮询数据的代码片段。

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);<br/>
    }
}

这是我正在设置的 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

不确定我在这里遗漏了什么。

Answer 1

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