Add ServiceBusProcessorAsyncClient (async message processor) (#46564)#49851
Draft
EldertGrootenboer wants to merge 16 commits into
Draft
Add ServiceBusProcessorAsyncClient (async message processor) (#46564)#49851EldertGrootenboer wants to merge 16 commits into
EldertGrootenboer wants to merge 16 commits into
Conversation
- 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 Azure#46564
|
Azure Pipelines: Successfully started running 1 pipeline(s). 34 pipeline(s) were filtered out due to trigger conditions. There may be pipelines that require an authorized user to comment /azp run to run. |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new reactive message-processing client, ServiceBusProcessorAsyncClient, to complement the existing synchronous ServiceBusProcessorClient. It enables non-blocking message and error handlers (Function<..., Mono<Void>>) while retaining the processor’s concurrency management, auto-recovery, and lifecycle semantics.
Changes:
- Added
ServiceBusProcessorAsyncClientwith reactive start/stop and best-effort drain-on-close behavior. - Added
processorAsync()/sessionProcessorAsync()builder entry points and async processor builder types inServiceBusClientBuilder. - Added unit tests and updated the Service Bus library CHANGELOG to document the new API.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessorAsyncClient.java | New reactive processor implementation (dispatch, auto-settlement, restart, drain-on-close). |
| sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusClientBuilder.java | New builder entry points and builder types for constructing the async processor clients. |
| sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusProcessorAsyncClientTest.java | New unit tests validating dispatch, settlement behavior, restart, draining, and lifecycle. |
| sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md | Documents the newly added async processor client feature. |
- 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.
… 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.
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Summary
Adds
ServiceBusProcessorAsyncClient, an asynchronous variant ofServiceBusProcessorClientwhose message and error handlers are reactive(
Function<..., Mono<Void>>) rather than blockingConsumer<T>. This letsapplications perform I/O-bound work in the handler without blocking a thread,
while keeping the processor's auto-recovery, concurrency management, and
lifecycle.
Resolves #46564.
What's new
ServiceBusProcessorAsyncClient— native reactive dispatch overServiceBusReceiverAsyncClient, bounded bymaxConcurrentCalls(no threadpinned per in-flight handler during I/O).
Monosuccess, abandons onhandler
Monoerror (opt out viadisableAutoComplete()).close()honoringdrainTimeout.Mono<Void> start()/stop(), blockingclose().ServiceBusClientBuilder.processorAsync()andsessionProcessorAsync(),returning
ServiceBusProcessorAsyncClientBuilder/ServiceBusSessionProcessorAsyncClientBuilder; build viabuildProcessorAsyncClient(). Full setter parity with the synchronousprocessor builder.
Cross-language alignment
.NET (
ServiceBusProcessor,Func<..., Task>) and JS(
processMessage: ... => Promise<void>) already accept async handlers; the Javaprocessor was the only one forcing a synchronous handler. This closes that gap
for the processor family.
Testing
disableAutoComplete, receiver-error to restart, completing-stream toexactly-one-restart, drain-on-close, PEEK_LOCK-skip vs
RECEIVE_AND_DELETE-process during drain (both directions), concurrency bounded
(upper and lower), stop then start resume, session dispatch + settlement,
in-band error context, and
getIdentifier()lifecycle.settlement, graceful drain).
revapiconfirms the change is purely additive.Notes for reviewers
processor emits) is intentionally deferred to a follow-up: it requires the
reactive tracer context-propagation API and does not affect the public
surface. The receiver's "receive" span is still emitted.
leaves draft.