Kafka使用者不在控制台中打印旧消息

Question

@EnableKafka 
@Configuration
public class KafkaConfig {
    
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id6");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        
        return new DefaultKafkaConsumerFactory<>(config);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> concurrentKafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
    
}

java

@Component
public class KafkaConsumer {

    @KafkaListener(topics = "test", groupId = "group_id6")
    public void consume(String message) {
        System.out.println("message: " + message);
    }
}

我正在使用控制中心来检查这个主题的消费者，并跟踪正在消费的数据。在运行这个应用程序时，它与Kafka和所有分区都连接得很好，我可以在控制中心看到所有的数据都被提取了，但在我的java控制台中没有打印任何数据。但是我注意到，在向Kafka发送一些新数据时，它会在java控制台中打印出来（即，在运行我的消费者后将新数据发送给Kafka）。它应该是这样的吗？还是我做错了什么？根据我的理解，Kafka主题中的旧消息也应该在控制台中打印出来。

Answer 1

在现有消费者组的情况下，当消费者再次启动时，它将从它在偏移量（或位置）中停止的地方继续，因为该偏移量由Kafka和/或Zookeeper存储。

另一方面，如果在一个现有主题中启动了一个新的消费者组，那么就没有偏移存储。在这种情况下，偏移量要么从主题的开头开始，要么从主题的结尾开始。

基本上，您应该创建一个新的使用者组并将其配置为从最小的偏移量开始，或者重置当前使用者组的偏移量。