KafkaConsumer 从头开始​​阅读主题

Question

我想从主题开始消费。我已将属性“AUTO_OFFSET_RESET_CONFIG”设置为最早，但不知何故仍未从头开始读取。

如果我遗漏了什么，有什么想法吗？我每次都在创建一个新的消费者群体。

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory
            = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);

    return factory;
}

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

private Map<String, Object> getConsumerConfigs(boolean isEmbedded) {
       Map<String, Object> props = new HashMap<>();
       props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, isEmbedded ? embeddedBootstrapServers : bootstrapServers);
       props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId + "temp");
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
       props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

       props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMillis);
       props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, heartbeatIntervalMillis);

       return props;
}

Answer 1

该属性仅适用于从未消费过的新消费者组。使用ConsumerSeekAware message listener，您可以为每个分配的主题/分区调用seekToBeginning。