如何为 onFailure 事件（Spring，Kafka）设置超时？答案

【问题标题】：How to set timeout for onFailure event (Spring, Kafka)?如何为 onFailure 事件（Spring，Kafka）设置超时？
【发布时间】：2018-09-04 08:54:25
【问题描述】：

我正在尝试在 Spring MVC 中实现向 Kafka 发送消息的异步 REST 方法。一切正常，但是当服务器不可用时，onFailure 事件会处理很长时间。例如，如何将 ListenableFuture 中的响应时间限制为三秒。

这是我的代码：

@Autowired
KafkaTemplate<String, String> kafkaTemplate;

@Value("${spring.kafka.topic}")
String topic;

@RequestMapping("/test")
DeferredResult<ResponseEntity<?>> test(
        @RequestParam(value = "message") String message
) {

    DeferredResult<ResponseEntity<?>> deferredResult = new DeferredResult<>();
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, "testKey", message);

    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

        @Override
        public void onSuccess(SendResult<String, String> sendResult) {
            ResponseEntity<String> responseEntity = new ResponseEntity<>("SUCCESS", HttpStatus.OK);
            deferredResult.setResult(responseEntity);
        }

        @Override
        public void onFailure(Throwable ex) {
            ResponseEntity<String> responseEntity = new ResponseEntity<>("FAILURE", HttpStatus.OK);
            deferredResult.setResult(responseEntity);
        }

    });

    return deferredResult;
}

我尝试使用 Kafka 的 REQUEST_TIMEOUT_MS_CONFIG 属性和 ListenableFuture 的 .get(long timeout, TimeUnit unit) 方法，但没有得到想要的结果。

【问题讨论】：

get() 应该在超时后失败；在那种情况下发生了什么？
@GaryRussell 如果我在return 之前添加future.get(1000, TimeUnit.MILLISECONDS);，我会在60 秒内收到以下错误。如果没有这个 get-method，我会在 60 秒内得到正确的“FAILURE”响应。 HTTP Status 500 - Request processing failed; nested exception is java.util.concurrent.ExecutionException: org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
看我的回答；你可以减少max.block.ms。

标签： java spring spring-mvc apache-kafka spring-kafka

【解决方案1】：

这是因为生产者阻塞了 60 秒（默认情况下）。

见max.block.ms in the KafkaDocumentation for producer configuration。

max.block.ms 配置控制 KafkaProducer.send() 和 KafkaProducer.partitionsFor() 将阻塞多长时间。这些方法可以被阻塞，因为缓冲区已满或元数据不可用。用户提供的序列化程序或分区程序中的阻塞将不计入此超时。

【讨论】：

非常感谢，这正是我所需要的！