Spring 是否允许为 KafkaTemplate 发送方法配置重试和恢复机制？

Question

尝试构建在执行kafkaTemplate.send() 方法期间可能发生的错误列表：

• 与序列化过程相关的错误；
• 一些网络问题或代理已关闭；
• 经纪人方面的一些技术问题，例如未收到经纪人的确认等。

现在我需要找到一种方法来以正确的方式处理所有可能的错误：

根据业务需求：如有异常我需要做以下事情：

• 重试3次；
• 如果所有 3 次重试都失败 - 记录相应的消息。

我发现配置属性spring.kafka.producer.retries 可用，我相信它正是我需要的。

但是我可以配置恢复方法（当所有重试失败时将执行的方法）？

Answer 1

spring.kafka.producer.retries 可能不是您想要的。 这个自动配置属性直接映射到ConsumerConfig：

map.from(this::getRetries).to(properties.in(ProducerConfig.RETRIES_CONFIG));

然后我们去阅读 ProducerConfig.RETRIES_CONFIG 属性的文档：

private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
        + " Note that this retry is no different than if the client resent the record upon receiving the error."
        + " Allowing retries without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
        + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
        + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
        + " failed before the number of retries has been exhausted if the timeout configured by"
        + " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
        + " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
        + " retry behavior.";

如您所见，spring-retry 完全不参与该过程，所有重试都直接在 Kafka 客户端及其KafkaProducer 基础架构内完成。

虽然这还不是全部。关注KafkaProducer.send()合约：

Future<RecordMetadata> send(ProducerRecord<K, V> record);

它返回一个Future。如果我们更仔细地看一下实现，我们会看到有一个同步部分——主题元数据请求和序列化，以及为异步发送到 Kafka 代理的批处理排队。提到的ProducerConfig.RETRIES_CONFIG 仅对Sender.completeBatch() 有效。

我相信当这些内部重试用尽时，Future 已完成并出现错误。因此，您可能应该考虑在KafkaTemplate 周围的服务方法中手动使用RetryTemplate，以便能够控制在当前调用中真正同步和阻塞的元数据和序列化的重试（和恢复）。您也可以通过重试在该方法中控制实际发送，但如果您调用 Future.get() 来阻止它在发送时来自 Kafka 客户端的响应或错误。