You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Adds ServiceBusProcessorAsyncClient for applications that compose message
processing with non-blocking I/O. Message and error handlers return Mono<Void>, so the processor tracks downstream work through completion
instead of treating callback return as the end of processing.
Behavior
The processor handles queues and subscriptions, including sessions through the
V2 asynchronous receiver path. maxConcurrentCalls bounds active handlers
across sessions. Automatic settlement completes a successful handler or
abandons a failed handler; disableAutoComplete() lets the handler return
reactive settlement as part of its work. Per-message renewal ends when the
full non-session processing operation completes, fails or is cancelled.
start() and stop() are cold operations. Stopping pauses dispatch while
accepted work can still settle; restarting resumes the retained receive
subscription. Blocking close() drains accepted handlers before disposing
receivers, up to drainTimeout. Session idle rollover waits for pending work.
Processing spans include error handling and abandonment. Terminal receive
streams recover through the ten-second monitor. Forwarded sends inherit the
processing span across asynchronous boundaries.
API
ServiceBusClientBuilder.processorAsync() and sessionProcessorAsync()
return builders with reactive message and error handlers. buildProcessorAsyncClient() creates the processor.
ServiceBusReceivedMessageAsyncContext exposes the message, entity path, and
namespace, plus cold complete(), abandon(), defer(), and deadLetter()
operations, each with an options overload.
What changed
Area
Changes
Processor and builders
Reactive dispatch, bounded concurrency, recovery, V2 sessions, and drain-aware lifecycle
Async message context
Reactive settlement and message metadata
Receivers, senders and instrumentation
Retained receive subscriptions, processing spans through settlement, and process-parent propagation when forwarding
Samples and documentation
Runnable seeded queue/session example, usage guidance, and changelog
Tests and fixtures
Context, processor, receiver, tracing, and live integration coverage
Testing
Before the renewal-lifetime correction, native clean mvn verify passes
833 executed cases across 76 suites, with 317 skipped and zero failures or
errors. Package checks, including coverage, API compatibility, Checkstyle,
SpotBugs, snippets and Spotless, pass with serial JUnit execution. The 22
focused regressions cover stop subscription during recovery cleanup,
forwarding across scheduler hops and all three send forms, explicit-parent
precedence, header preservation and repeated subscription. A clean Java 11
run passes all 26 parallel lifecycle and tracing controls.
Earlier live verification comprises a passing 22-case processor run and a
passing four-case tracing run, with zero failures, errors or skips. Only the
tracing fixture changes between those runs; production code and processor
tests are unchanged. Processor coverage includes session/non-session queues
and subscriptions, renewal, close drain, redelivery, receive-and-delete,
stop/resume, disabled renewal and session rollover. Tracing checks first-send
ancestry and single, iterable and batch forwarding after a scheduler hop,
including exact parent IDs, span counts, wire-preserved headers and settlement.
The renewal-lifetime correction passes all 185 affected Java 11 processor,
receiver and renewal cases. Six regressions cover unsettled successful and
failed handlers and drain cancellation through the real V1/V2 receiver.
Both new broker cases pass: queue and subscription handlers renew during
processing, fail without settling, then receive and complete redelivery
within 30 seconds while the processor remains open. Earlier passing suites
are retained rather than repeated.
Review
This adds public API. APIView and maintainer review remain outstanding.
- New ServiceBusProcessorAsyncClient with reactive message and error handlers
(Function<..., Mono<Void>>), dispatch bounded by maxConcurrentCalls
- Async auto-settlement (complete on success, abandon on handler error),
monitor-based auto-recovery, and a best-effort draining close()
- New processorAsync() / sessionProcessorAsync() factory methods and the
ServiceBusProcessorAsyncClientBuilder / ServiceBusSessionProcessorAsyncClientBuilder
builders (build via buildProcessorAsyncClient())
- 16 unit tests covering dispatch, settlement, error handling, drain, recovery,
concurrency, and the PEEK_LOCK-skip vs RECEIVE_AND_DELETE-process drain asymmetry
- CHANGELOG entry for 7.18.0-beta.3
Fixes#46564
- Validate entity/connection inputs at build time in both async builders
(buildProcessorAsyncClient), so a misconfiguration fails fast as documented
instead of being deferred to start()
- Separate settlement errors from handler errors in dispatchMessage: a
complete() failure is now reported to the error handler with its own error
source (e.g. COMPLETE) and no longer misclassified as USER_CALLBACK or
followed by a spurious abandon()
- Mark the test close-threads as daemon so a failed assertion cannot keep the
test JVM alive and hang the suite
- Add completeFailureIsReportedAndNotAbandoned test (asserts COMPLETE source
and no abandon)
- Validate maxConcurrentCalls in both constructors (fail-fast, matching the
builder message) so an invalid 0/negative value cannot reach the Reactor
flatMap concurrency
- Document that close()/stop() must not be called from a message or error
handler (it would deadlock the drain); schedule shutdown from a separate thread
- Add constructorRejectsInvalidMaxConcurrentCalls test
- In dispatchMessage, skip a redeliverable PEEK_LOCK message during close()
drain via a fast-path early return BEFORE incrementing activeHandlerCount, so
churn from skipped messages under sustained load cannot keep the drain above
zero and delay shutdown. The post-increment re-check is kept to handle the
check-then-act race (matches the sync MessagePump pattern).
- Fix the CI javadoc:jar failure: {@link Mono#block()} pulled in reactor''s
@nullable (a jsr305 @nonnull(when=When.MAYBE) nickname) that javadoc cannot
resolve and treats as a fatal warning. Use {@code block()} instead. Also
removed the javadoc-only builder imports, fully-qualified the nested-builder
@see/@link references, and switched the Schedulers reference to {@code}.
- Fix "queuename" -> "queue name" typo in both async builder subQueue javadoc.
- Gate auto-settlement on the receive mode: in RECEIVE_AND_DELETE the broker
removes the message on delivery, so complete()/abandon() are meaningless and
would fail. dispatchMessage now settles only when the receiver is PEEK_LOCK
(settleMessages = autoComplete && PEEK_LOCK), matching the receiver builder
which disables auto-complete for RECEIVE_AND_DELETE.
- Test: default the mock receiver to PEEK_LOCK; assert never().complete()/
abandon() in the RECEIVE_AND_DELETE drain test.
- The processMessage javadoc implied strictly-serial requesting; clarify that
up to maxConcurrentCalls messages are processed concurrently and a new
message is requested as each in-flight handler''s Mono terminates.
- Wrap the disableAutoComplete() {@link} across two lines to stay within the
120-char checkstyle LineLength limit.
- "non session-enabled" -> "non-session-enabled".
… limit
- The async session processor applies maxConcurrentCalls as a single flatMap
concurrency bound across all active sessions, not per session. Update the
javadoc to describe the actual behavior rather than the sync per-session model.
- Under sustained RECEIVE_AND_DELETE traffic the receiver keeps delivering while
the drain waits, so the in-flight set may not reach zero before the drain
timeout; handlers cancelled at timeout drop already-deleted messages. Document
this best-effort limitation; a future revision may decouple handler execution
from the receiver subscription to bound it.
- restartMessageReceiver() calls a blocking client close(); it was invoked
directly from the receive subscription''s onError/onComplete callbacks, which
can block a reactive thread. Schedule it on Schedulers.boundedElastic()
(matching the monitor path); the restart guard makes a stale schedule a no-op.
- Wrap the ServiceBusReceivedMessageContext construction so a synchronous throw
after the activeHandlerCount increment still decrements it, preventing a
permanent count > 0 that would block close() until the drain timeout.
- Schedule the restart via Mono.fromRunnable(...).subscribeOn(boundedElastic())
so the blocking restart runs off the reactive thread without an ignored
Scheduler.schedule() Disposable.
- receivesMessagesAndAutoCompletes and inBandErrorContextInvokesErrorHandler
asserted a callback should not run via Mono.error(new AssertionError(...)),
but the processor swallows callback errors so those never failed the test.
Record the invocation in an AtomicReference/AtomicBoolean and assert it stays
unset after close.
- stop() javadoc said in-flight handlers are not interrupted, but stop()
disposes the receive subscription and so cancels in-flight flatMap handlers.
State that it does not wait and directs callers to close() for best-effort
draining.
- close() javadoc listed stop() alongside close() as blocking-until-drain. Only
close() blocks/drains; stop() disposes without draining. Clarify that calling
stop() from a handler cancels the invoking handler.
- subQueue/subscriptionName/topicName javadoc used @see with a full sentence
as the label (e.g. "@see #queueName A queue name or #topicName ..."), which
renders as a broken "See Also" and leaves #topicName unlinked.
- subQueue: move the note into the description with proper {@link} refs and
keep the valid @see SubQueue.
- subscriptionName/topicName: the description already states the companion
field is required, so drop the redundant malformed @see.
- Scoped to the two new async builders; checkstyle + javadoc clean.
- Integrate main into the existing async processor feature branch
- Complete V2 sessions, reactive settlement, draining, and tracing
- Add seeded samples and verified native and live coverage
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Azure Pipelines:
Successfully started running 1 pipeline(s).
35 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.
Adds a reactive Service Bus processor to the Azure SDK for Java, addressing #46564 by tracking asynchronous handler completion rather than callback return.
Changes:
Adds async processor builders and message contexts with reactive settlement.
- Pace receiver recovery through the existing monitor
- Normalize disabled session renewal and schedule restart cleanup safely
- Trace error handling and abandonment through completion
- Cover all four review findings with focused regressions
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
The new public processor API and custom AMQP lifecycle require maintainer/APIView review, and reported live tests do not cover the final cleanup and recovery changes.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
A critical race allows an old receiver’s terminal callback to mark a healthy replacement as terminated, causing the next monitor tick to close it and cancel active handlers.
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
The new public processor APIs and session, recovery, and drain lifecycles still require maintainer and APIView sign-off, plus real-AMQP validation of the final lifecycle race changes.
No open issues 🎉
2 previously missed issues
In code that hasn't changed since last review
Schedule stop cleanup off Reactor event-loop threads
Subscribing to stop() on a Reactor event-loop thread can block that thread while automatic recovery holds the same lock during synchronous receiver cleanup, even without concurrent lifecycle calls from the application. Schedule this operation on bounded elastic, as start() already does, so cleanup cannot block the subscribing thread. Add a regression that holds recovery cleanup while subscribing to stop() from a non-blocking scheduler.
Propagate Reactor context into asynchronous sender tracing
A handler returning sender.sendMessage(new ServiceBusMessage(...)) starts a disconnected trace with SDK-only OpenTelemetry configuration: the sender's message and send instrumentation do not read the process span placed in Reactor context here. Make sender tracing consume that context without holding a thread-local scope across asynchronous signals. Add a forwarding-handler test that verifies trace and parent IDs, including after a scheduler hop.
- Offload stop subscriptions during receiver recovery.
- Preserve processing parents across send and batch operations.
- Correct tracing flags and settle the first-send test message.
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
stop() now schedules its subscription on bounded elastic, matching start().
Forwarded sends and asynchronously created batches retain the processing span as their parent across scheduler hops, while preserving explicit message parents and existing headers.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
A failed, unsettled non-session handler using disableAutoComplete() leaves lock renewal running after processing ends, delaying redelivery for up to the configured renewal duration.
- Release non-session renewal after full processing termination.
- Preserve renewal during drain and asynchronous error handling.
- Cover unsettled V1/V2 handlers and broker redelivery.
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
The new public processor APIs introduce intertwined session concurrency, renewal ownership, and stop/drain contracts that still require APIView and Service Bus maintainer review.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes Azure/azure-sdk-for-java#46564 - Support a ServiceBusProcessorAsyncClient.
Adds
ServiceBusProcessorAsyncClientfor applications that compose messageprocessing with non-blocking I/O. Message and error handlers return
Mono<Void>, so the processor tracks downstream work through completioninstead of treating callback return as the end of processing.
Behavior
The processor handles queues and subscriptions, including sessions through the
V2 asynchronous receiver path.
maxConcurrentCallsbounds active handlersacross sessions. Automatic settlement completes a successful handler or
abandons a failed handler;
disableAutoComplete()lets the handler returnreactive settlement as part of its work. Per-message renewal ends when the
full non-session processing operation completes, fails or is cancelled.
start()andstop()are cold operations. Stopping pauses dispatch whileaccepted work can still settle; restarting resumes the retained receive
subscription. Blocking
close()drains accepted handlers before disposingreceivers, up to
drainTimeout. Session idle rollover waits for pending work.Processing spans include error handling and abandonment. Terminal receive
streams recover through the ten-second monitor. Forwarded sends inherit the
processing span across asynchronous boundaries.
API
ServiceBusClientBuilder.processorAsync()andsessionProcessorAsync()return builders with reactive message and error handlers.
buildProcessorAsyncClient()creates the processor.Processor lifecycle:
ServiceBusReceivedMessageAsyncContextexposes the message, entity path, andnamespace, plus cold
complete(),abandon(),defer(), anddeadLetter()operations, each with an options overload.
What changed
Testing
Before the renewal-lifetime correction, native clean
mvn verifypasses833 executed cases across 76 suites, with 317 skipped and zero failures or
errors. Package checks, including coverage, API compatibility, Checkstyle,
SpotBugs, snippets and Spotless, pass with serial JUnit execution. The 22
focused regressions cover stop subscription during recovery cleanup,
forwarding across scheduler hops and all three send forms, explicit-parent
precedence, header preservation and repeated subscription. A clean Java 11
run passes all 26 parallel lifecycle and tracing controls.
Earlier live verification comprises a passing 22-case processor run and a
passing four-case tracing run, with zero failures, errors or skips. Only the
tracing fixture changes between those runs; production code and processor
tests are unchanged. Processor coverage includes session/non-session queues
and subscriptions, renewal, close drain, redelivery, receive-and-delete,
stop/resume, disabled renewal and session rollover. Tracing checks first-send
ancestry and single, iterable and batch forwarding after a scheduler hop,
including exact parent IDs, span counts, wire-preserved headers and settlement.
The renewal-lifetime correction passes all 185 affected Java 11 processor,
receiver and renewal cases. Six regressions cover unsettled successful and
failed handlers and drain cancellation through the real V1/V2 receiver.
Both new broker cases pass: queue and subscription handlers renew during
processing, fail without settling, then receive and complete redelivery
within 30 seconds while the processor remains open. Earlier passing suites
are retained rather than repeated.
Review
This adds public API. APIView and maintainer review remain outstanding.