配置failure-strategy=ignore后SmallRye Reactive Messaging异常停收消息

预期行为 #

根据Quarkus Kafka相关文档，failure-strategy=ignore 的行为为：记录失败日志，继续处理流程并提交处理失败的记录偏移量。

配置内容 #

mp.messaging.incoming.words-in.topic=words
mp.messaging.outgoing.words-out.topic=words
mp.messaging.incoming.words-in.auto.offset.reset=earliest
mp.messaging.incoming.words-in.commit-strategy=throttled
mp.messaging.incoming.words-in.failure-strategy=ignore

测试代码 #

@Incoming("words-in")
@NonBlocking
public Uni<Void> storeToDB(Message<String> message) {
    Log.info("Received publish direction");
    return test(message).onFailure().invoke(throwable -> {
        System.out.println("NACK!!!");
        System.out.println("NACK!!!");
        System.out.println("NACK!!!");
        System.out.println("NACK!!!---");
        message.nack(throwable);
    });
}

public Uni<Void> test(Message<String> message) {
    return Uni.createFrom().failure(new IllegalAccessError("TEST"));
}

执行输出日志 #

2022-09-30 13:50:22,268 INFO  [com.sim.mes.PublishModuleProcessor] (vert.x-eventloop-thread-6) Received publish direction
NACK!!!
NACK!!!
NACK!!!
NACK!!!---
2022-09-30 13:50:22,270 WARN  [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-6) SRMSG18204: A message sent to channel `words-in` has been nacked, ignored failure is: TEST.
2022-09-30 13:50:22,271 ERROR [io.sma.rea.mes.provider] (vert.x-eventloop-thread-6) SRMSG00201: Error caught while processing a message: java.lang.IllegalAccessError: TEST
    at com.acme.messaging.PublishModuleProcessor.test(PublishModuleProcessor.java:75)
    at com.acme.messaging.PublishModuleProcessor_Subclass.test$$superforward1(Unknown Source)
    at com.acme.messaging.PublishModuleProcessor_Subclass$$function$$75.apply(Unknown Source)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:53)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(Unknown Source)
    at io.quarkus.arc.impl.InterceptorInvocation.invoke(InvocationInterceptor.java:41)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:40)
    at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
    at com.acme.messaging.PublishModuleProcessor_Subclass.test(Unknown Source)
    at com.acme.messaging.PublishModuleProcessor.storeToDB(PublishModuleProcessor.java:65)
    at com.acme.messaging.PublishModuleProcessor_Subclass.storeToDB$$superforward1(Unknown Source)
    at com.acme.messaging.PublishModuleProcessor_Subclass$$function$$74.apply(Unknown Source)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:53)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(Unknown Source)
    at io.quarkus.arc.impl.InterceptorInvocation.invoke(InvocationInterceptor.java:41)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:40)
    at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
    at com.acme.messaging.PublishModuleProcessor_Subclass.storeToDB(Unknown Source)
    at com.acme.messaging.PublishModuleProcessor_ClientProxy.storeToDB(Unknown Source)
    at com.acme.messaging.PublishModuleProcessor_SmallRyeMessagingInvoker_storeToDB_308e8bd532ece24c38471c9208b94b5b4fa0f2f7.invoke(Unknown Source)
    at io.smallrye.reactive.messaging.providers.AbstractMediator.invoke(AbstractMediator.java:95)
    at io.smallrye.reactive.messaging.providers.AbstractMediator.lambda$invokeOnMessageContext$1(AbstractMediator.java:103)
    at io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata.lambda$invokeOnMessageContext$0(LocalContextMetadata.java:34)
    at io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata.lambda$invokeOnMessageContext$2(LocalContextMetadata.java:55)
    at io.smallrye.context.impl.wrappers.SlowContextualConsumer.accept(SlowContextualConsumer.java:21)
    at io.smallrye.mutiny.operators.uni.builders.UniCreateWithEmitter.subscribe(UniCreateWithEmitter.java:22)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni.subscribe(UniOnItemTransformToUni.java:25)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap.subscribe(UniOnItemOrFailureFlatMap.java:27)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.request(UniToMultiPublisher.java:74)
    at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapInner.onSubscribe(MultiFlatMapOp.java:601)
    at io.smallrye.mutiny.converters.uni.UniToMultiPublisher.subscribe(UniToMultiPublisher.java:26)
    at io.smallrye.mutiny.groups.MultiCreate$1.subscribe(MultiCreate.java:163)
    at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapMainSubscriber.onItem(MultiFlatMapOp.java:193)
    at io.smallrye.mutiny.operators.multi.MultiMapOp$MapProcessor.onItem(MultiMapOp.java:50)
    at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
    at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:99)
    at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)
    at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:85)
    at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.onItem(MultiOperatorProcessor.java:100)
    at io.smallrye.reactive.messaging.providers.locals.ContextDecorator$ContextMulti$ContextProcessor.lambda$onItem$1(ContextDecorator.java:78)
    at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
    at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
    at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
    at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:503)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:833)

疑问 #

配置failure-strategy=ignore后，预期处理失败的消息时会提交偏移量并继续处理后续消息，但实际发生异常后，队列后续消息不再被处理，必须重启服务才能恢复。请问我的理解是否有误，该如何解决？

内容的提问来源于stack exchange，提问作者Nicholas Loke