springspring-bootazureazureservicebusservicebus

Error when I try to receive messages from Azure Service Bus


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.


Solution

  • 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.