如何设置onFailure事件(春季，卡夫卡)的超时时间? [英] How to set timeout for onFailure event (Spring, Kafka)?

问题描述

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

I'm trying to implement an asynchronous REST method of sending a message to Kafka in Spring MVC. Everything works, but when the server is unavailable, the onFailure event is processed for a long time. How to limit the response time in ListenableFuture for example to three seconds.

这是我的代码:

@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(长时间超时，TimeUnit单位)方法，但没有得到想要的结果.

I tried to use a REQUEST_TIMEOUT_MS_CONFIG property of Kafka and .get(long timeout, TimeUnit unit) method of ListenableFuture but havn't got desired result.

推荐答案

这是因为生产者会阻塞60秒(默认情况下).

That's because the producer blocks for 60 seconds (by default).

有关生产者配置的信息，请参见KafkaDocumentation中的 max.block.ms .

See max.block.ms in the KafkaDocumentation for producer configuration.

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

max.block.ms The configuration controls how long KafkaProducer.send() and KafkaProducer.partitionsFor() will block.These methods can be blocked either because the buffer is full or metadata unavailable.Blocking in the user-supplied serializers or partitioner will not be counted against this timeout.

这篇关于如何设置onFailure事件(春季，卡夫卡)的超时时间?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！