Kafka 无法处理所有消息 - Java Spring Boot

Question

我有一个 Spring Boot 应用程序（Spring 版本 2.2.2.RELEASE），我在其中配置了 Kafka 消费者，它处理来自 Kafka 的数据并服务于多个 Web 套接字。订阅 kafka 成功，但并非选定 Kafka 主题的所有消息都由消费者处理。很少有消息被延迟，很少有消息被错过。但是生产者正在发送完全确保的数据。下面我分享了我使用过的配置属性。

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    final String BOOTSTRAP_SERVERS = kafkaBootstrapServer;
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    return new DefaultKafkaConsumerFactory<>(props);
}

我是否缺少任何配置？

Answer 1

对于新的消费者（从不提交 group.id 的偏移量），您必须将 AUTO_OFFSET_RESET 设置为 earliest 以避免丢失主题中的任何现有记录（默认为 latest）。