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

Question

生产者发送消息到一个有四个分区的主题。我们有一个消费者在消费来自这个主题的消息。应用程序在工作日一直运行周末例外：它不会在周末期间调用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天）。

在您的情况下，由于消费者组周末Rest，因此会重置offset。

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