在使用reactor kafka进行并发处理的同时，重试使用消息

Question

我有一个项目，在这个项目中，我使用来自传入主题的消息进行处理，并将它们发送到传出主题。现在考虑一下，由于某些基础结构问题导致发送消息时出错，传出主题已关闭。在这种情况下，我将提交到最后一条成功的消息（成功处理并发送到传出主题），并从发送失败的地方重试消费。

这可以在“非反应性”kafka中使用ack. nack（index，0017）轻松实现，代码如下所示：

@KafkaListener(
        autoStartup = "false",
        topics = "incoming-topic",
        containerFactory = "myListenerContainerFactory",
        groupId = "test-consumer"
)
public void consume(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
    for (int i = 0; i < consumerRecords.size(); i++) {
        ConsumerRecord<String, String> record = consumerRecords.get(i);
        try {
            JsonObject recordNode = gson.fromJson(record.value(), JsonObject.class);
            CompletableFuture<String> future1 = transformationService.transform1(recordNode);
            CompletableFuture<String> future2 = transformationService.transform2(recordNode);
            CompletableFuture.allOf(future1, future2).join();

            outgoingTopicProducer.sendMessage(record, recordNode); // method that calls producer
            ack.acknowledge();
        } catch (ApiException e) { // ApiException thrown from producer
            logger.error("Kafka exception occurred while sending message, blocking queue... ");
            ack.nack(i, 2000); // this enables to read same message again after 2 secs
        } catch (Exception e) { // ignore any other exception like JsonProcessingException
            logger.error("error occurred while processing message. Partition: {}, Offset: {}, Value: {}",
                    record.partition(),
                    record.offset(),
                    record.value());
            logger.error("exception: " + e.getMessage());
        }
    }
}

现在，为了获得更好的性能和处理背压，我想让这个代码变得被动。我还想并行处理每个分区。通过查看Reactorkafka文档，我得出了以下代码：

private final Scheduler scheduler = Schedulers.newBoundedElastic(60, Integer.MAX_VALUE, "sample", 60, true);
private final KafkaSender<String, String> sender = KafkaSender.create(SenderOptions.create(getProducerConfig()));
@EventListener(ApplicationReadyEvent.class)
public void receiverFlux() {
    KafkaReceiver.create(getReceiverOptions())
        .receive()
        .groupBy(m -> m.receiverOffset().topicPartition())
        .flatMap(partitionFlux ->
            partitionFlux.publishOn(scheduler)
                    .<SenderRecord<String, String, ReceiverOffset>>handle((record, sink) -> {
                        JsonObject recordNode = gson.fromJson(record.value(), JsonObject.class);

                        CompletableFuture<String> future1 = transformationService.transform1(recordNode);
                        Mono<String> mono1 = Mono.fromFuture(future1)
                                .subscribeOn(Schedulers.parallel())
                                .doFinally(sig -> {
                                    if (sig == SignalType.CANCEL) {
                                        future1.cancel(true);
                                    }
                                });

                        CompletableFuture<String> future2 = transformationService.transform2(recordNode);
                        Mono<String> mono2 = Mono.fromFuture(future2)
                                .subscribeOn(Schedulers.parallel())
                                .doFinally(sig -> {
                                    if (sig == SignalType.CANCEL) {
                                        future2.cancel(true);
                                    }
                                });

                        try {
                            Mono.when(mono1, mono2).block();
                            sink.next(SenderRecord.create(new ProducerRecord<>("outgoing-topic", record.partition(), record.timestamp(), record.key(), recordNode.toString(), record.headers()),
                                    record.receiverOffset()));
                        } catch (Exception e) {
                            sink.error(e);
                        }
                    })
                    .onErrorResume(e -> {
                        logger.error("error occurred wile processing record");
                        logger.error("exception: " + e.getMessage());
                        return Mono.empty();
                    })
                    .as(sender::send)
                    .doOnNext(record -> {
                        logger.info("ack offset");
                        record.correlationMetadata().acknowledge();
                    })
                    .doOnError(e -> {
                        logger.error("error occurred while sending message");
                        logger.error("exception: " + e.getMessage());
                    })
                    .retryWhen(Retry.backoff(10, Duration.ofSeconds(2)))

        ).onErrorResume(e -> {
            logger.error("Restart consumer ....");
            return Mono.empty();
        })
        .repeat()
        .subscribe();
}

为了测试这一点，我将传出主题的名称改为不正确的主题，并得到以下日志:

{"@message":"request(1)"}
{"@message":"[Producer clientId=clientId] Error while fetching metadata with correlation id 1 : {incorrect-topic=TOPIC_AUTHORIZATION_FAILED}"}
{"@message":"[Producer clientId=clientId] Topic authorization failed for topics [incorrect-topic]"}
{"@message":"Sender failed","@fields":{"level":"ERR","throwable":{"message":"Not authorized to access topics: [incorrect-topic]"}}}
{"@message":"error occurred while sending message"}
{"@message":"exception: Not authorized to access topics: [incorrect-topic]"}
{"@message":"Scheduler worker in group main failed with an uncaught exception","@fields{"level":"ERR","throwable":{"stackTrace":[{"classLoaderName":"app","moduleName":null,"moduleVersion":null,"methodName":"runAsync","fileName":"FluxPublishOn.java","lineNumber":1004,"nativeMethod":false,"className":"reactor.core.publisher.FluxPublishOn$PublishOnConditionalSubscriber"},{"classLoaderName":"app","moduleName":null,"moduleVersion":null,"methodName":"run","fileName":"FluxPublishOn.java","lineNumber":1067,"nativeMethod":false,"className":"reactor.core.publisher.FluxPublishOn$PublishOnConditionalSubscriber"},{"classLoaderName":"app","moduleName":null,"moduleVersion":null,"methodName":"call","fileName":"WorkerTask.java","lineNumber":84,"nativeMethod":false,"className":"reactor.core.scheduler.WorkerTask"},{"classLoaderName":"app","moduleName":null,"moduleVersion":null,"methodName":"call","fileName":"WorkerTask.java","lineNumber":37,"nativeMethod":false,"className":"reactor.core.scheduler.WorkerTask"},{"classLoaderName":null,"moduleName":"java.base","moduleVersion":"15","methodName":"run","fileName":"FutureTask.java","lineNumber":264,"nativeMethod":false,"className":"java.util.concurrent.FutureTask"},{"classLoaderName":null,"moduleName":"java.base","moduleVersion":"15","methodName":"run","fileName":"ScheduledThreadPoolExecutor.java","lineNumber":304,"nativeMethod":false,"className":"java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask"},{"classLoaderName":null,"moduleName":"java.base","moduleVersion":"15","methodName":"runWorker","fileName":"ThreadPoolExecutor.java","lineNumber":1130,"nativeMethod":false,"className":"java.util.concurrent.ThreadPoolExecutor"},{"classLoaderName":null,"moduleName":"java.base","moduleVersion":"15","methodName":"run","fileName":"ThreadPoolExecutor.java","lineNumber":630,"nativeMethod":false,"className":"java.util.concurrent.ThreadPoolExecutor$Worker"},{"classLoaderName":null,"moduleName":"java.base","moduleVersion":"15","methodName":"run","fileName":"Thread.java","lineNumber":832,"nativeMethod":false,"className":"java.lang.Thread"}],"message":"Cannot invoke \"java.util.Queue.isEmpty()\" because \"q\" is null","suppressed":[],"localizedMessage":"Cannot invoke \"java.util.Queue.isEmpty()\" because \"q\" is null"}}}

因此，观察结果是，此代码不会在失败的情况下重试发送消息，当新消息到来时，它会读取新消息，删除最后一条失败的消息，导致该消息丢失。优先级是我不想丢失任何消息。

请帮助我实现这些功能。欢迎任何其他有助于提高性能的建议。

Answer 1

我能够通过在ReceiverRecord通量周围使用defer运算符来解决这个问题。

Flux.defer(() -> KafkaReceiver.create(getReceiverOptions())
    .receive())

代码的其余部分不需要任何更改。