JavaSpring Kafka模板生产者在代理重启时丢失消息

Question

我将sping-boot（2.1.6.RELEASE）与sping-kafka（2.2.7.RELEASE）一起使用，并且我使用KafkaTemplate向我的kafka集群发送消息。但是有时（通常是当我重新启动kafka代理或进行重新平衡时），我在发送消息时会看到这样的错误：

org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

由于默认的Kafka生产者配置，我期望发送失败重试，但他们没有。默认Kafka生成器配置:

retries: 2147483647  (https://kafka.apache.org/documentation/#retries)
acks: 1               (https://kafka.apache.org/documentation/#acks)

我的配置是这样的：

@Bean
    public Map<String, Object> producerConfigs()
    {
        // See https://kafka.apache.org/documentation/#producerconfigs for more properties
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
        return props;
    }

    @Bean
    public ProducerFactory<Long, String> producerFactory()
    {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<Long, String> kafkaTemplate(KafkaTemplateProducerListener<Long, String> kafkaTemplateProducerListener,
                                                     ProducerFactory<Long, String> producerFactory)
    {
        KafkaTemplate<Long, String> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setProducerListener(kafkaTemplateProducerListener);
        return kafkaTemplate;
    }

我发出这样的信息:

kafkaTemplate.send(topicName, key, body);

我在互联网上搜索了所有的人，每个人都说这种带有重试和acks的配置必须有效，但它没有。我错过了什么？

谢啦

Answer 1

经过一段时间的调试，我找到了解决方案：

props.put(ProducerConfig.ACKS_CONFIG, "all");

有关此属性的更多信息：https://kafka.apache.org/documentation/#acks
非常好的博客，展示了在kafka中可能丢失消息的不同场景：

• 第1部分：https://jack-vanlightly.com/blog/2018/9/14/how-to-lose-messages-on-a-kafka-cluster-part1
• 第2部分：https://jack-vanlightly.com/blog/2018/9/18/how-to-lose-messages-on-a-kafka-cluster-part-2

旁注-根据这个答案，我发现如果您不想在关机时丢失消息，最好使用这个：

@PreDestroy
public void flush()
{
    kafkaTemplate.flush();
}