[BUG] ServiceBusReactorReceiver NPE "this.isDisposed is null": overridden closeAsync called from ReactorReceiver constructor when connection shutdown signal already emitted #51163

Describe the bug
ServiceBusReactorReceiver throws NullPointerException: ... "this.isDisposed" is null when a receive link is created on an AMQP connection that has already emitted its shutdown signal.

This is the same issue as #46325, which was closed without a root cause. The root cause:

1. ServiceBusReactorReceiver.<init> calls super(...) (ServiceBusReactorReceiver.java:76). Per Java initialization order, the subclass field initializers have not run yet, so isDisposed (line 58) is null and isV2 is false.
2. The ReactorReceiver constructor subscribes to amqpConnection.getShutdownSignals() (ReactorReceiver.java:201-204) and calls closeAsync(...) in the flatMap.
3. If the shutdown signal was already emitted, SinkOneMulticast replays it synchronously inside subscribe(), while we are still inside the parent constructor.
4. closeAsync(String, ErrorCondition) dispatches to the ServiceBusReactorReceiver override. isV2 is still false, so it falls through to isDisposed.getAndSet(true) (line 199). isDisposed is null, so it throws an NPE.

isDisposed is never reassigned, as noted in #46325. It is null because an overridable method is called from the parent constructor before the subclass is initialized.

The error ends up in Reactor's onErrorDropped. The processor recovers on a new connection, so the impact is errorlog/alert noise, so message is not loss.

Exception or Stack Trace

reactor.core.publisher.Operators : Operator called default onErrorDropped
java.lang.NullPointerException: Cannot invoke "java.util.concurrent.atomic.AtomicBoolean.getAndSet(boolean)" because "this.isDisposed" is null
    at com.azure.messaging.servicebus.implementation.ServiceBusReactorReceiver.closeAsync(ServiceBusReactorReceiver.java:199)
    at com.azure.core.amqp.implementation.ReactorReceiver.lambda$new$12(ReactorReceiver.java:203)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:389)
    at reactor.core.publisher.Operators$MonoInnerProducerBase.complete(Operators.java:2827)
    at reactor.core.publisher.SinkOneMulticast.subscribe(SinkOneMulticast.java:101)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8888)
    at reactor.core.publisher.Flux.subscribeWith(Flux.java:9009)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8853)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8777)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8695)
    at com.azure.core.amqp.implementation.ReactorReceiver.<init>(ReactorReceiver.java:204)
    at com.azure.messaging.servicebus.implementation.ServiceBusReactorReceiver.<init>(ServiceBusReactorReceiver.java:76)
    at com.azure.messaging.servicebus.implementation.ServiceBusAmqpLinkProvider.createReceiveLink(ServiceBusAmqpLinkProvider.java:45)
    at com.azure.core.amqp.implementation.ConsumerFactory.createConsumer(ConsumerFactory.java:64)
    at com.azure.core.amqp.implementation.ReactorSession.getSubscription(ReactorSession.java:672)
    at com.azure.core.amqp.implementation.ReactorSession.lambda$createConsumer$2(ReactorSession.java:448)
    at java.util.concurrent.ConcurrentHashMap.compute
    at com.azure.core.amqp.implementation.ReactorSession.lambda$createConsumer$1(ReactorSession.java:434)
    at com.azure.core.amqp.implementation.handler.DispatchHandler.onTimerTask(DispatchHandler.java:32)
    at com.azure.core.amqp.implementation.ReactorDispatcher$WorkScheduler.run(ReactorDispatcher.java:219)
    at org.apache.qpid.proton.reactor.impl.SelectableImpl.readable(SelectableImpl.java:118)
    at org.apache.qpid.proton.reactor.impl.IOHandler.handleQuiesced(IOHandler.java:61)
    at org.apache.qpid.proton.reactor.impl.IOHandler.onUnhandled(IOHandler.java:398)
    at org.apache.qpid.proton.engine.BaseHandler.onReactorQuiesced(BaseHandler.java:87)
    at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:206)
    at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:292)
    at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:91)
    at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:72)
    at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
    at java.util.concurrent.FutureTask.run
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run
    at java.util.concurrent.ThreadPoolExecutor.runWorker
    at java.util.concurrent.ThreadPoolExecutor$Worker.run
    at java.lang.Thread.run

To Reproduce
This is a race between AMQP connection shutdown and receive-link creation, so it is not reproducible on demand in a running app. It occurred about 60s after application start, with ~20 ServiceBusProcessorClient instances (queues and topic subscriptions, PEEK_LOCK, auto-complete disabled) running across 2 namespaces.

Here are Conceptual Steps :

1. A ServiceBusProcessorClient is running.
2. The underlying AMQP connection shuts down (service/network side), and its shutdown signal is emitted.
3. A receive-link creation task that is already queued on the ReactorDispatcher runs ReactorSession.createConsumer on that connection.
4. new ServiceBusReactorReceiver(...) throws the NPE above.

Code Snippet
The relevant SDK code (azure-messaging-servicebus 7.17.19 / azure-core-amqp 2.12.0):

// ServiceBusReactorReceiver.java
private final AtomicBoolean isDisposed = new AtomicBoolean();   // line 58: runs only AFTER super(...) returns

public ServiceBusReactorReceiver(AmqpConnection connection, String entityPath, Receiver receiver, ...) {
    super(connection, entityPath, receiver, handler, tokenManager, dispatcher, retryOptions,   // li
        new AmqpMetricsProvider(null, connection.getFullyQualifiedNamespace(), entityPath));
    ...
}

@Override
protected Mono<Void> closeAsync(String message, ErrorCondition errorCondition) {
    if (isV2) {                                   // false during super(...)
        return super.closeAsync(message, errorCondition);
    }
    if (isDisposed.getAndSet(true)) {             // line 199: NPE, isDisposed == null
        return super.getIsClosedMono();
    }
    ...
}

// ReactorReceiver.java (constructor)
amqpConnection.getShutdownSignals().flatMap(signal -> {          // line 201
    logger.verbose("Shutdown signal received.");
    return closeAsync("Connection shutdown.", null);             // line 203: dispatches to subclass
}).subscribe());                                                 // line 204: replays synchronously

Expected behavior
Creating a receive link on an already-shut-down connection should close the link cleanly (or fail wiithout an NPE from a partially constructed object.

Screenshots
N/A

Setup (please complete the following information):

• OS: Linux container (Kubernetes / AKS)
• IDE: IntelliJ IDEA
• Library/Libraries:
  • com.azure:azure-messaging-servicebus:7.17.19
  • com.azure:azure-core-amqp:2.12.0
  • com.azure:azure-core:1.58.1
  • com.azure.spring:spring-cloud-azure-starter:7.4.0
• Java version: 25 (Eclipse Temurin 25.0.5 JRE)
• App Server/Environment: Spring Boot app in Kubernetes
• Frameworks: Spring Boot 4.1.2, Kotlin 2.4.10

Additional context

• Same issue as [BUG] NPE #46325 (reported on 7.17.12, closed without a fix).
• The code is unchanged in the latest versions I checked: azure-messaging-servicebus 7.17.20 and 7.19.1, and azure-core-amqp 2.13.0.
• The NPE is reported via Operators.onErrorDropped (in [BUG] NPE #46325 it surfaced as ErrorCallbackNotImplemented), so it shows up as an error in monitoring (Sentry) even though the processor self-heals.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Bug Description Added
• Repro Steps Added
• Setup information Added