I'm trying to receive messages from Azure Service Bus in my SpringBoot application, but I'm facing some issues. This is my code:
Consumer<ServiceBusReceivedMessageContext> processMessage = messageContext -> {
try {
System.out.println(messageContext.getMessage().getMessageId());
// other message processing code
messageContext.complete();
} catch (Exception ex) {
messageContext.abandon();
}
};
Consumer<ServiceBusErrorContext> processError = errorContext -> {
System.err.println("Error occurred while receiving message: " + errorContext.getException());
};
ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
.connectionString(<<CONNECTION_STRING>>)
.processor()
.queueName("test")
.disableAutoComplete()
.receiveMode(PEEK_LOCK)
.processMessage(processMessage)
.processError(processError)
.disableAutoComplete()
.buildProcessorClient();
Basically, I'm using the same code of this example but I'm receiving this error when I run my application:
java.lang.NoSuchFieldError: CLIENT_RECEIVER_IDENTIFIER at com.azure.messaging.servicebus.implementation.ServiceBusReactorSession.createConsumer(ServiceBusReactorSession.java:185) ~[azure-messaging-servicebus-7.11.0.jar:7.11.0] at com.azure.messaging.servicebus.implementation.ServiceBusReactorSession.createConsumer(ServiceBusReactorSession.java:99) ~[azure-messaging-servicebus-7.11.0.jar:7.11.0] at com.azure.messaging.servicebus.implementation.ServiceBusReactorAmqpConnection.lambda$createReceiveLink$3(ServiceBusReactorAmqpConnection.java:183) ~[azure-messaging-servicebus-7.11.0.jar:7.11.0] at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:125) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:292) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:236) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:203) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.SerializedSubscriber.onComplete(SerializedSubscriber.java:146) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.SerializedSubscriber.onComplete(SerializedSubscriber.java:146) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.onComplete(FluxTimeout.java:234) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoNext$NextSubscriber.onComplete(MonoNext.java:102) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:83) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:118) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:877) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:965) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxReplay$ReplaySubscriber.onNext(FluxReplay.java:1344) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxDistinctUntilChanged$DistinctUntilChangedSubscriber.tryOnNext(FluxDistinctUntilChanged.java:149) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxDistinctUntilChanged$DistinctUntilChangedSubscriber.onNext(FluxDistinctUntilChanged.java:102) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:877) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:965) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.ReplayProcessor.tryEmitNext(ReplayProcessor.java:508) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27) ~[reactor-core-3.4.22.jar:3.4.22] at com.azure.core.amqp.implementation.handler.Handler.onNext(Handler.java:89) ~[azure-core-amqp-2.5.2.jar:2.5.2] at com.azure.core.amqp.implementation.handler.SessionHandler.onSessionRemoteOpen(SessionHandler.java:87) ~[azure-core-amqp-2.5.2.jar:2.5.2] at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:146) ~[proton-j-0.33.6.jar:na] at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108) ~[proton-j-0.33.6.jar:na] at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324) ~[proton-j-0.33.6.jar:na] at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291) ~[proton-j-0.33.6.jar:na] at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:91) ~[azure-core-amqp-2.5.2.jar:2.5.2] at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) ~[reactor-core-3.4.22.jar:3.4.22] at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) ~[reactor-core-3.4.22.jar:3.4.22] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na] at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na] at java.base/java.lang.Thread.run(Thread.java:834) ~[na:an]
It's weird because when I try to use the peekMessage function using the ServiceBusReceiverAsyncClient it works, but when I try to use the receiveMessages or this approach using the ProcessorClient, it throws the error. I didn't find anything about this error, I don't know if it's a property that I have to set.
Apparently it was related to te package version, previously I was using the following:
<dependency>
<groupId>com.azure</groupId>
<artifactId>azure-messaging-servicebus</artifactId>
<version>7.10.1</version>
</dependency>
when I rollback to a previous version (7.10.1), it started to work properly, I don't know if there's a compatibility issue, but now it's working properly.
Thank you.