Reactor Kafka生产者-无法重试

Question

我已经使用（reactor.kafka.sender.KafkaSender）和Reactor Kafka（用于Kafka的功能Java API）创建了KafkaProducer。使用以下生产者配置，

max.block.ms = 8000
request.timeout.ms= 4000
retries = 3
retry.backoff.ms = 2000
max.in.flight.requests.per.connection = 512
acks = all

[当我尝试将记录发布到无效主题时，我收到超时异常

org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 8000 ms

这符合预期。但是我已经配置了不会发生的重试。我假设max.block.ms / request.timeout.ms失效后，将在每个retry.backoff.ms之后重试，直到metadata.max.age.ms或retries用尽。仅供参考，代码：

    String topic = "order/";
    int count = 1;
    Flux<SenderRecord<String, Event, EventInfo>> source = Flux.range(1, count).map(x -> {
      Event event = new Event();
      return SenderRecord.create(
            new ProducerRecord<String, Event>(topic, event.getX(),
                event), event.getEvent());
    });
    kafkaSender.send(source).subscribe(x -> System.out.println(x));
    kafkaSender.close();

• 配置是否正确以启用重试？
• request.timeout.ms / max.block.ms之后何时重试？
• 上面的代码需要进行哪些更改以允许重试？

Answer 1

我相信您还应该设置“ delivery.timeout.ms”

请参阅此处的文档：https://docs.confluent.io/current/installation/configuration/producer-configs.html#retries

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 max.in.flight.requests.per.connection 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 delivery.timeout.ms expires first before successful acknowledgement. Users should generally prefer to leave this config unset and instead use delivery.timeout.ms to control retry behavior.