春季-卡夫卡制片人在所有经纪人停机时重试 [英] Spring-Kafka Producer Retry when all brokers are down

问题描述

我正在使用Springboot2.3.5.RELEASE和SpringKafka 2.6.3。我正在尝试做一个简单的卡夫卡生产者重试POC，这应该会导致生产者重试时，经纪人关闭或如果有一个异常抛出之前，消息被发送到经纪人。
以下生产者配置用于启用重试的幂等生产者。

// Producer configuration
@Bean
public Map<String, Object> producerConfigs() {
   Map<String, Object> props = new HashMap<>();
   props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
   props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
   props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
   props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "3");
   props.put(ProducerConfig.RETRIES_CONFIG, "10");
   props.put(ProducerConfig.ACKS_CONFIG, "all");
}
@Bean
public ProducerFactory<String, Object> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}
... other configs

我知道，仅当Broker向Producer发送ack时出现暂时性错误时，Kafka Producer重试才起作用。因此，如果代理甚至在消息发送之前就已关闭，则上述重试将不起作用。 因此，我尝试引入@retry with Spring-Kafka，不幸的是，我也无法让它工作。有谁知道如何解决这个问题，因为这似乎是一种常见的用法，比如如果经纪人倒闭或如果出现网络故障，我不希望我的制片人停止工作

@Transaction
@Retryable(value = Exception.class, maxAttemptsExpression = "10",
        backoff = @Backoff(delayExpression = "1000"))
public void sendTestMessage(String name, String number) throws RuntimeException {
    TestObject testObj = new TestObject(name + ": " + String.valueOf(number),
            Double.valueOf(Math.random() * 10000).longValue(),
            "Blah" + String.valueOf(Math.random() * 10));
    log.info("Sending sendTestMessage {}", name); 
    throwSomeException();
    // sends the message
    kafkaTemplate.send(MyConstants.SomeTOPIC, testObj);     
}  
// Throwing mock exception
public void throwSomeException() throws RuntimeException {
        log.info("Throwing mock exception {}");
        throw new RuntimeException("Mock Exception thrown while sending the 
        sendTestMessage method");
}   
@Recover
public void recover(Exception e, TestObject testObj) { 
    log.info("Recovered exception thrown  mock exception {} with payload as {}", e, testObj);
}

是否可以重试以解决上述问题。我很难相信这样的功能在默认情况下不是内置的。

推荐答案

我解决了这个问题，我发布这篇文章是为了让其他人受益。 我的配置类中缺少以下内容

@Configuration
@EnableRetry <-- this was missing
public class ProducerConfig {
  ...

添加上述内容后，它工作正常。

这篇关于春季-卡夫卡制片人在所有经纪人停机时重试的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！