java - Spring kafka消费者不尊重自动偏移重置=最新

Question

我们有 1 个 Kafka 主题和 1 个分区:

从 spring boot kafka 消费者那里看到一个相当奇怪的行为。 Spring kafka消费者在重新启动时总是从主题的开头开始消费。 我已经配置了spring kafka监听器如下

卡夫卡监听器:

@KafkaListener(topics = "${application.kafkaInputTopic}", groupId = "${spring.kafka.consumer.group-id}")
public void listen(String message) {
    log.debug("SG message received. Parsing...");
    TransmissionMessage transmissionMessage;
    SGTransmission transmission = parseMessage(message);
    //Porcess Transmission......
}

消费者配置和Spring消费者容器 Autowiring bean

@Resource
public Environment env;

@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(consumerFactory());
   // I know this isnt right, should be run in 1 thread as there isonly 
   //partition in the topic
    factory.setConcurrency(10);
    factory.getContainerProperties().setPollTimeout(3000);

    factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);

    factory.getContainerProperties().setSyncCommits(true);
    return factory;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> propsMap = new HashMap();
    propsMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, env.getProperty(Constants.SPRING_KAFKA_SECURITY_PROTOCOL));
    propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty(Constants.SPRING_KAFKA_BOOTSTRAP_SERVERS));
    propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty(Constants.SPRING_KAFKA_GROUP_ID));
    return propsMap;
}

Spring 应用yaml

kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS}
    properties:
    consumer:
        # If this consumer does not have an offset yet, start at latest offset.
        # Be careful with `earliest`, this will use the first (available) offset in the topic, which is most likely not what you want.
        auto-offset-reset: latest
        group-id: ${KAFKA_GROUP_ID}

每次消费者崩溃并重新启动时，所有消息都会从头开始读取。正如您在 application.yaml 中看到的那样，情况不应该是这样的

auto-offset-reset: latest

代理端或消费者端是否可能存在一些我可能忽略的其他配置，这导致消费者每次重新启动时都从头开始读取？

Answer 1

您必须以某种方式提交初始偏移量，也许是在完成此配置之前。

factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);

这意味着您有责任提交偏移量。

使用AckMode.BATCH(默认)或AckMode.RECORD。

或者，使用 kafka-consumer-groups CLI 工具删除当前提交的偏移量(您也可以使用相同的工具列出当前偏移量)。

或者使用组的 UUID 来每次获取一个新组。

编辑

您还可以让监听器类实现 ConsumerSeekAware 并在 onPartitionsAssigned() 中调用 callback.seekToEnd(partitions)。