ReactorKafka制作人-无法重试

Question

我已经使用reactor-kafka（kafka的一个功能性Java API）创建了一个KafkaProducer（reactor.kafka.sender.KafkaSender）。使用以下生产者配置，

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/请求。暂停。ms已过期，每次重试后都会重试。退后。ms直到元数据。最大年龄。ms或重试次数已用尽。仅供参考，代码：

    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();

• 启用重试的配置是否正确




Answer 1

我相信你也应该设置"delivery.timeout.ms"

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

将值设置为大于零将导致客户端重新发送发送失败并可能出现瞬时错误的任何记录。请注意，此重试与客户端在收到错误时重新发送记录没有什么不同。允许重试而不将max.in.flight.requests.per.connection设置为1可能会更改记录的顺序，因为如果将两个批次发送到单个分区，并且第一批失败并重试，但第二批成功，那么第二批中的记录可能会首先出现。另外请注意，如果delivery.timeout.ms配置的超时在成功确认之前首先过期，那么在重试次数用尽之前，产生请求将失败。用户通常应该倾向于不设置此配置，而是使用delivery.timeout.ms来控制重试行为。