Repository navigation
[BUG] azure-core-amqp: ActiveClientTokenManager interrupts an in-progress token refresh ("IO Sink was interrupted before reactor closed") #50922
Description
Describe the bug
ActiveClientTokenManager schedules refreshes with switchOnNext(durationSource.map(Flux::interval)). The next
refresh duration is emitted from the subscriber of the current refresh (on the CBS reply / reactor thread), possibly
while the current interval tick is still running. switchOnNext cancels the running interval, and cancelling a periodic
task on Schedulers.parallel() calls Future.cancel(true), interrupting the thread performing the refresh. If that
thread is writing to the reactor's IO pipe (ReactorDispatcher.signalWorkQueue), the write fails with
ClosedByInterruptException, the pipe is closed and the whole connection is reset.
Exception or Stack Trace
java.nio.channels.ClosedByInterruptException
at java.base/java.nio.channels.spi.AbstractInterruptibleChannel.end(...)
at java.base/sun.nio.ch.SinkChannelImpl.write(...)
at com.azure.core.amqp.implementation.ReactorDispatcher.signalWorkQueue(...)
...
IO Sink was interrupted before reactor closed
Expected behavior
Scheduling the next token refresh never interrupts work of the current one.
To Reproduce
Long-running producer with token-based auth; intermittent connection resets around token refresh. Deterministic unit
repro: see ActiveClientTokenManagerTest in the linked PR.
Setup
- Java 17, azure-core-amqp 2.12.1 (same code on main as of 2026-10-21)
Activity
github-actions commentedon Oct 22, 2026
🎯 Agentic Issue Triage
Summary:
ActiveClientTokenManagerin azure-core-amqp usesswitchOnNextoverFlux::interval, so scheduling the next refresh can cancel and interrupt the in-progress refresh. This can causeClosedByInterruptExceptioninReactorDispatcher.signalWorkQueueand a connection reset.📋 Issue Details
Package:
com.azure:azure-core-amqp2.12.1Affected API:
ActiveClientTokenManager(token refresh scheduling),ReactorDispatcher.signalWorkQueueScenarios:
Root ask: Scheduling the next refresh must not interrupt the current one (the author references a linked PR with a unit test).
🔎 Debugging / Reproduction Notes
The cause is plausible: cancelling a periodic task on
Schedulers.parallel()interrupts its thread.Suggested investigation steps:
ActiveClientTokenManagerscheduling and theswitchOnNextusage.ActiveClientTokenManagerTestrepro from the linked PR.switchOnNextDelayError, orpublishOn/subscribeOnwith non-interrupting cancellation).🏷️ Label Confidence
Client— SDK code behavior bug.Azure.Core.AMQP— the package is azure-core-amqp.👥 Owner Routing
# ServiceLabel: %Azure.Core.AMQP(line 68) — the service label matches.@conniey%Mgmt(line 945) did not match because the issue is not Mgmt.github-actions commentedon Oct 22, 2026
Thank you for your feedback. Tagging and routing to the team member(s) best able to assist.