Spring WebFlux中Reactor-Kafka遇RecordDeserializationException时如何手动提交偏移量？

问题背景 #

在使用Spring WebFlux搭配Reactor-Kafka接收器时，遇到RecordDeserializationException异常时，虽能从异常中提取对应的TopicPartition和偏移量，但由于ReceiverOffset采用私有实现，无法手动创建实例执行提交操作，导致无法跳过坏消息并推进消费偏移量。现有核心代码如下：

reactiveKafkaReceiver
        .receiveBatch()
        .onErrorResume(e -> {
            RecordDeserializationException rde = (RecordDeserializationException) e;
            TopicPartition topicPartition = rde.topicPartition();
            long offset = rde.offset();
            // how can I commit this offset?
            return Flux.empty();
        })
        .delayUntil(flux -> flux
                .collectList()
                .delayUntil(this::process)
                .doOnNext(records -> records.forEach(record -> record.receiverOffset()
                        .commit()
                        .subscribeOn(Schedulers.boundedElastic())
                        .subscribe())))
        .retryWhen((Retry.backoff(3, Duration.ofSeconds(2)).transientErrors(true)))
        .repeat()
        .subscribe();

可行解决方案 #

方法1：通过Reactor-Kafka的ConsumerAccessor操作底层KafkaConsumer #

Reactor-Kafka的ReceiverOptions提供了consumerAccessor()方法，可直接获取底层KafkaConsumer实例，调用其原生提交API完成偏移量提交。

代码示例

// 提前初始化并持有ReceiverOptions实例（确保全局复用）
private final ReceiverOptions<String, Object> receiverOptions;

// 消费逻辑调整
reactiveKafkaReceiver
        .receiveBatch()
        .onErrorResume(e -> {
            if (e instanceof RecordDeserializationException rde) {
                TopicPartition tp = rde.topicPartition();
                long failedOffset = rde.offset();
                // Kafka提交的是「下一个要消费的偏移量」，因此需要+1
                Map<TopicPartition, OffsetAndMetadata> offsetMap = Collections.singletonMap(
                        tp, new OffsetAndMetadata(failedOffset + 1)
                );
                // 非阻塞执行提交（避免阻塞WebFlux事件循环）
                Mono.fromRunnable(() -> receiverOptions.consumerAccessor().consumer().commitSync(offsetMap))
                    .subscribeOn(Schedulers.boundedElastic())
                    .subscribe(
                        () -> {},
                        commitErr -> log.error("提交偏移量失败", commitErr)
                    );
            }
            return Flux.empty();
        })
        .delayUntil(flux -> flux
                .collectList()
                .delayUntil(this::process)
                .doOnNext(records -> records.forEach(record -> record.receiverOffset()
                        .commit()
                        .subscribeOn(Schedulers.boundedElastic())
                        .subscribe())))
        .retryWhen(Retry.backoff(3, Duration.ofSeconds(2)).transientErrors(true))
        .repeat()
        .subscribe();

方法2：自定义反序列化器提前捕获异常 #

通过自定义反序列化器包裹原生实现，确保反序列化异常被正确封装为RecordDeserializationException，后续仍可复用方法1的提交逻辑。

代码示例

public class ErrorHandlingDeserializer<T> implements Deserializer<T> {
    private final Deserializer<T> delegate;
    private final Logger log = LoggerFactory.getLogger(ErrorHandlingDeserializer.class);

    public ErrorHandlingDeserializer(Deserializer<T> delegate) {
        this.delegate = delegate;
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return delegate.deserialize(topic, data);
        } catch (Exception e) {
            log.error("反序列化消息失败，topic: {}", topic, e);
            throw new RecordDeserializationException(topic, null, data, e);
        }
    }

    @Override
    public T deserialize(String topic, Headers headers, byte[] data) {
        try {
            return delegate.deserialize(topic, headers, data);
        } catch (Exception e) {
            log.error("反序列化带header的消息失败，topic: {}", topic, e);
            throw new RecordDeserializationException(topic, null, data, e);
        }
    }
}

配置ReceiverOptions时替换原生反序列化器：

ReceiverOptions<String, Object> receiverOptions = ReceiverOptions.<String, Object>create(kafkaProps)
        .keyDeserializer(new ErrorHandlingDeserializer<>(new StringDeserializer()))
        .valueDeserializer(new ErrorHandlingDeserializer<>(new JsonDeserializer<>(Object.class)));

关键注意事项 #

• 偏移量计算：必须提交failedOffset + 1，因为Kafka的偏移量提交逻辑是标记「下一个要消费的消息位置」，否则会重复消费坏消息。
• 非阻塞提交：在WebFlux的非阻塞场景下，务必将commitSync()/commitAsync()放到单独的线程池执行（如Schedulers.boundedElastic()），避免阻塞事件循环。
• 异常边界：批量消费时需注意区分单个消息异常和整个批次异常，避免误提交整个批次的偏移量。

内容的提问来源于stack exchange，提问作者Mikhail Geyer