Kafka 消费者正在重用来自主题的消息

Question

由于某种原因，我们 Java 应用程序中的 Kafka 消费者正在重复使用来自主题的消息。

生产者向一个有四个分区的主题发送消息。 我们有一个消费者使用来自该主题的消息。应用程序在工作日异常周末一直运行：它不会在周末调用 poll 方法。
消费者配置：自动提交，自动提交时间为 5s（默认）。

应用程序一直运行良好，直到周日它再次恢复调用 poll 方法。我们看到从该主题轮询了数百万条消息。消费者基本上是在轮询来自主题的所有消息。将新偏移量与周末停止之前的偏移量进行比较时。新的偏移量要小得多，就像将所有四个分区都重置为非常低的数字。

我们不知道消费者端发生了什么，因为它没有调用 poll 方法，所以没有打印日志消息。我们检查了 kafka 服务器日志，但我们什么也没找到。

以前有人见过吗？消费者会被错误配置吗？

    <spring.kafka.version>1.1.2.RELEASE</spring.kafka.version>
     ... 
  <bean id="defaultKafkaConsumer"
        class="org.apache.kafka.clients.consumer.KafkaConsumer">
        <constructor-arg>  
            <map>
                <entry key="bootstrap.servers"  value="${kafka.bootstrap.servers}"></entry>
                <entry key="max.block.ms"  value="5000"></entry>
                <entry key="group.id"  value="kafkaconnect.tca"></entry>
                <entry key="auto.offset.reset" value="earliest"></entry>
                <entry key="key.deserializer"  value="org.apache.kafka.common.serialization.StringDeserializer"></entry>
                <entry key="value.deserializer"  value="org.apache.kafka.common.serialization.StringDeserializer"></entry>
            </map>  
        </constructor-arg>  
    </bean>

    getKafkaConsumer().subscribe(Arrays.asList(getKafkaTopic()));
            // set up the polling task
            handler = timer.scheduleAtFixedRate(new Runnable() {
                public void run() {
                    try {
                        processPoll();
                    } catch (Throwable t) {
                        LOG.error(String.format("error processing poll for inet: %s, details: %s - %s", getId(), t.getMessage(), t.getCause()), t);
                    }
                }
            }, 3, 3, TimeUnit.MILLISECONDS);



   processPoll() Method: destination will not be ready during the weekend.
   try {
            if (!isDestinationReady()) {
                if (destinationIgnoreCnt++ ==0) {
                    LOG.warn(String.format("outbound destination session is not ready - trying: %s/%s",destinationIgnoreCnt,destinationwaitingloop));
                } else if ((destinationIgnoreCnt++ % destinationwaitingloop) == 0) {
                    LOG.warn(String.format("outbound destination session is not ready - trying %s/%s", destinationIgnoreCnt,destinationwaitingloop));
                    destinationIgnoreCnt = 1;
                }
                messageIgnoreCnt = 0;
                return;
            }
            if(!isDestinationOpen()) {
                if (destinationIgnoreCnt++ ==0) {
                    LOG.error(String.format("outbound destination is not opended - trying:%s/%s.", destinationIgnoreCnt,destinationwaitingloop) );
                } else if ((destinationIgnoreCnt++ % destinationwaitingloop) == 0) {
                    LOG.error(String.format("outbound destination is not opended - trying %s/%s.", destinationIgnoreCnt,destinationwaitingloop));
                    destinationIgnoreCnt = 1;
                }
                messageIgnoreCnt = 0;
                return;
            }

            if (messageIgnoreCnt++ == 0) {
                LOG.info(String.format("kafka poller started. poll interval %s wait: %s", pollingInterval, 60000));
            } else if ((messageIgnoreCnt++ % 30) == 0) {// approximately 30mins
                LOG.info(String.format("kafka poller started. poll interval %s wait %s", pollingInterval, 60000));
                messageIgnoreCnt = 1;
            }

            if (getKafkaConsumer() == null) {
                LOG.critical("KafkaListener consumer is null");
                return;
            }

            ConsumerRecords<String, String> records = getKafkaConsumer().poll(60000);
            if (records == null || records.isEmpty()) {
                LOG.debug("zero records received from Kafka");
                return;
            }
            for (ConsumerRecord<String, String> record : records) {
                LOG.info(String.format("consuming from topic =  %s ", record.toString()));
                try {
                    String jsonMsg = record.value();

                    DirectBatchRequest payload = JacksonUtilities.getObjectMapper().readValue(jsonMsg, DirectBatchRequest.class);

                    if (payload != null) {
                        LOG.info(String.format("Got it reportId:%s", payload.getDestinationId()));
                        if(payload.getDestinationId() == 0) {
                            LOG.info(String.format("Applying default destination desk:%s", defaultDeskId));
                            payload.setDestinationId(defaultDeskId);
                        }
                        List<RequestEntryType> requestEntryTypeList = ((StreamDirectRequest) payload).getRequestList();
                        LOG.info(String.format("Processing size: %s" , requestEntryTypeList.size()) );
                        processRequest((StreamDirectRequest) payload);  //async call
                        LOG.info(String.format("Processing size: %s sent to Steam" , requestEntryTypeList.size()) );    
                    }
                } catch (Throwable t) {
                    LOG.error(String.format("KafkaListener JSON%s conversion error %s", record, t.getMessage()));
                }

            }

        } catch (Throwable t) {
            LOG.error(String.format("KafkaListener exception %s", t.getMessage()));

        }

Answer 1

如果消费者组没有活动，Kafka 默认会删除offsets.retention.minutes 之后的偏移量。默认保留期为 1440 分钟（1 天）。

在您的情况下，由于消费者组周末关闭，因此重置了偏移量。

见https://kafka.apache.org/documentation/#brokerconfigs