Closing reactor connection and children. #19924

conniey · 2021-03-17T20:49:05Z

Closing reactor connection and children.
Adding asynchronous disposes so we wait for the connection/sessions/links to complete.
Updating *Processor usage to new reactor Sinks..
Performing remoteCredit checks in reactor dispatcher so that we don't get into a condition where in flux deliveries are being settled before we add credits.

Related: #18070

conniey · 2021-03-17T21:09:03Z

/azp run java - eventhubs - tests

azure-pipelines · 2021-03-17T21:09:15Z

Azure Pipelines successfully started running 1 pipeline(s).

Changing double hasStarted check.

… for each interval (for a Flux).

…ved.

… longer "connected".

…ssed as fast.

…sed.

…into conniey/messaging-factory

...ore/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AsyncAutoCloseable.java

...core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java

* Adding lock around creation of reactor. * Align with Track 1 by scheduling pending sends after operation timeout. Changing hasStarted check. * ReactorSession has reference to AmqpConnection (parent) to dispose. * SessionHandler passes along its remote state instead of Active. * Only sets errorCondition when it exists. * Emitting a connection when it is active. Also, adding timeout messages. * Connect AmqpConnection to dispose of child when shutdown signal received. * When endpointStates or shutdownSignals are received, the sender is no longer "connected". * ReactorSender: Rename to isClosed. Adding Mono. * ReactorReceiver: Adding async closing. * ReactorExecutor: Close scheduler when there is no more work. * ReactorSender: Change logging to verbose. * ReactorSession: Close session when it has completed all tasks. * Update connection to close when sessions are disposed. * Update Connection to only propagate more shutdown signals if it has not been signaled already. * Adding retry time to close ReactorExecutor. * Moves decodeMessage and remote credit into dispatcher.invoke method. * Do not use .distinct() for endpoint handlers. * Use boundedElastic in EventHubClient instead of deprecated elastic(). * Dispose sender and receiver when it is no longer authorized. * Using distinctUntilChanged() rather than keeping track of endpoints myself. * Add PartitionPump to release the scheduler when a partition is closed. * Use try/catch around dispose in case an exception occurs. * Adding shutdown signals to ReactorDispatcher * Adding AsyncAutoCloseable and using it to dispose Reactor classes. * Emitting exception when IOException thrown before reactor is closed. * Catching RejectedExecutionException in the case that the IO signal is closed, so we can close resources manually. * Removing unused scheduler. * Removing duplicated logging. * Fixing possible NPE between checking key and getting value. * Only take while AzureCredentialTokenManager is not disposed. * Checking for NPE when closing links. * Removing comment about elastic scheduler since it is not configurable.

srnagar · 2021-04-12T17:38:18Z

...ore/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AsyncAutoCloseable.java

+/**
+ * Asynchronous class to close resources.
+ */
+public interface AsyncAutoCloseable extends AutoCloseable {


Why do we want to extend AutoCloseable? Any class that wants to support both sync and async close method can implement both interfaces.

Jonathan had mentioned that it would be nice if our AsyncAutoCloseable works nicely with try-with-resources syntactic sugar.

srnagar · 2021-04-12T17:54:18Z

...core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java

+                if (isDisposed.getAndSet(true)) {
+                    logger.verbose("connectionId[{}] was already disposed. {}", connectionId, message);
+                } else {
+                    dispose(new AmqpShutdownSignal(false, false, message));


When an error occurs, we should inspect the AMQP error to ensure that the error was meant for this connection by comparing the connection id in the error with the current connection id. Otherwise, we'll dispose of a good connection when the error was for an older connection that is no longer active. We should do the same for session and link errors as well.

The ConnectionHandler is bound to a single connection, so it will always be the same one. When we create a proton-j Connection, we bind an instance of the ConnectionHandler to it. So each connection has its own connection handler instance.

srnagar · 2021-04-12T18:00:57Z

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java

+                    // When we encounter an error refreshing authorization results, close the receive link.
+                    final Mono<Void> operation =
+                        closeAsync(String.format(
+                            "connectionId[%s] linkName[%s] Token renewal failure. Disposing receive link.",
+                            amqpConnection.getId(), getLinkName()),
+                            new ErrorCondition(Symbol.getSymbol(AmqpErrorCondition.NOT_ALLOWED.getErrorCondition()),
+                                error.getMessage()));
+
+                    return operation.then(Mono.empty());


Can this be done in the subscribe's error handler instead of using onErrorResume? It is odd to have an empty error handler below.

I had that before but on an error, we want to output another async operation (ie. close the receiver) and the handlers only support synchronous callbacks, I moved it back to be a part of the chain.

srnagar · 2021-04-12T18:02:24Z

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java

            } catch (IOException e) {
-                logger.warning("Unable to schedule work to add more credits.", e);
+                sink.error(new RuntimeException(String.format(


Use UncheckedIOException instead when mapping IOException to a runtime exception.
https://docs.oracle.com/javase/8/docs/api/java/io/UncheckedIOException.html

Sweet. I've never heard of this.

srnagar · 2021-04-12T18:27:02Z

...ubs/src/main/java/com/azure/messaging/eventhubs/implementation/AmqpReceiveLinkProcessor.java

-                }));
+                next.receive()
+                    .onBackpressureBuffer(maxQueueSize, BufferOverflowStrategy.ERROR)
+                    .subscribe(message -> {


Should this subscribe have an error handler to dispose the link if there's an error?

No because the underlying deliverySink this fetches from never outputs an onError, it will do that via the endpointStates

* Adding lock around creation of reactor. * Align with Track 1 by scheduling pending sends after operation timeout. Changing hasStarted check. * ReactorSession has reference to AmqpConnection (parent) to dispose. * SessionHandler passes along its remote state instead of Active. * Only sets errorCondition when it exists. * Emitting a connection when it is active. Also, adding timeout messages. * Connect AmqpConnection to dispose of child when shutdown signal received. * When endpointStates or shutdownSignals are received, the sender is no longer "connected". * ReactorSender: Rename to isClosed. Adding Mono. * ReactorReceiver: Adding async closing. * ReactorExecutor: Close scheduler when there is no more work. * ReactorSender: Change logging to verbose. * ReactorSession: Close session when it has completed all tasks. * Update connection to close when sessions are disposed. * Update Connection to only propagate more shutdown signals if it has not been signaled already. * Adding retry time to close ReactorExecutor. * Moves decodeMessage and remote credit into dispatcher.invoke method. * Do not use .distinct() for endpoint handlers. * Use boundedElastic in EventHubClient instead of deprecated elastic(). * Dispose sender and receiver when it is no longer authorized. * Using distinctUntilChanged() rather than keeping track of endpoints myself. * Add PartitionPump to release the scheduler when a partition is closed. * Use try/catch around dispose in case an exception occurs. * Adding shutdown signals to ReactorDispatcher * Adding AsyncAutoCloseable and using it to dispose Reactor classes. * Emitting exception when IOException thrown before reactor is closed. * Catching RejectedExecutionException in the case that the IO signal is closed, so we can close resources manually. * Removing unused scheduler. * Removing duplicated logging. * Fixing possible NPE between checking key and getting value. * Only take while AzureCredentialTokenManager is not disposed. * Checking for NPE when closing links. * Removing comment about elastic scheduler since it is not configurable.

conniey self-assigned this Mar 17, 2021

ghost added Azure.Core azure-core Event Hubs Service Bus labels Mar 17, 2021

conniey force-pushed the conniey/messaging-factory branch 2 times, most recently from 08eb9ab to ce4a583 Compare March 17, 2021 21:05

conniey force-pushed the conniey/messaging-factory branch 3 times, most recently from 998f964 to 8d6ef0e Compare March 23, 2021 06:56

conniey changed the title ~~(WIP) Closing reactor connection and children.~~ Closing reactor connection and children. Mar 24, 2021

conniey added 18 commits March 23, 2021 19:58

Adding lock around creation of reactor.

988926d

Align with Track 1 by scheduling pending sends after operation timeout.

063be9c

Changing double hasStarted check.

Update RetryUtil to consider the first emission. Otherwise, it checks…

8441693

… for each interval (for a Flux).

ReactorSession has reference to AmqpConnection (parent) to dispose.

8670201

SessionHandler passes along its remote state instead of Active.

6d28673

Fix order of stop and process according to the docs.

c28a9c1

Only sets errorCondition when it exists.

bbffad4

Emitting a connection when it is active. Also, adding timeout messages.

a99928d

Add TODO comment in Receiver to close when we are unauthorized.

8d5f825

Connect AmqpConnection to dispose of child when shutdown signal recei…

c1fd05e

…ved.

Linking ReactorSender to use AmqpConnection.

584e825

When endpointStates or shutdownSignals are received, the sender is no…

dd5ba13

… longer "connected".

ReactorSender: Rename to isClosed. Adding Mono.

fe8f29d

ReactorReceiver: Adding async closing.

5dc178f

ReactorSender fixing tests and connection errors.

02ab253

ReactorReceiver adding support for onClose mono.

62f5023

ReactorExecutor: Close scheduler when there is no more work.

ed154ef

ReactorSender: Change logging to verbose.

645b358

conniey added 11 commits April 7, 2021 10:56

Only take while it is not disposed.

b699bd4

Checking for NPE when closing links.

d25b5b3

Create backpressure queue in the case that deliveries cannot be proce…

6ad5fbd

…ssed as fast.

Adding log to see additional credits being added.

d3a98f7

Updating error logs.

8c9881d

Fixing test breaks.

c034f18

Adding back backpressure

ad52aed

Do not recursively call remove.

b1cd78f

Fix imports

3c41608

Fixing possible NPE when subscribing first.

9bc17a1

Creating a new provider for send links in the case the reactor is clo…

72be7e6

…sed.

conniey mentioned this pull request Apr 8, 2021

[BUG] EventHub Consumer stops consuming messages until we restart #18070

Closed

3 tasks

conniey added 5 commits April 8, 2021 09:12

Merge branch 'master' of https://github.com/Azure/azure-sdk-for-java …

ba7f0b5

…into conniey/messaging-factory

Fix test breaks in Service Bus.

eff9e35

Use cold subscriber for messages.

6fe96ea

Removing comment about elastic scheduler since it is not configurable.

423afd0

Fix tests breaks in service bus.

bf241ee

hemanttanwar reviewed Apr 8, 2021

View reviewed changes

Adding documentation and fixing variable names.

7eebad1

conniey force-pushed the conniey/messaging-factory branch from 952c1f0 to 500ad8c Compare April 9, 2021 18:45

Fixing test.

45d9f8e

conniey force-pushed the conniey/messaging-factory branch from 500ad8c to 45d9f8e Compare April 9, 2021 19:06

hemanttanwar approved these changes Apr 12, 2021

View reviewed changes

conniey merged commit 6b66681 into master Apr 12, 2021

conniey deleted the conniey/messaging-factory branch April 12, 2021 16:31

srnagar reviewed Apr 12, 2021

View reviewed changes

conniey mentioned this pull request Apr 12, 2021

Returning UncheckedIOException instead of RuntimeException. #20568

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Closing reactor connection and children. #19924

Closing reactor connection and children. #19924

conniey commented Mar 17, 2021 •

edited

Loading

conniey commented Mar 17, 2021

azure-pipelines bot commented Mar 17, 2021

srnagar Apr 12, 2021

conniey Apr 12, 2021

srnagar Apr 12, 2021

conniey Apr 12, 2021

srnagar Apr 12, 2021

conniey Apr 12, 2021

srnagar Apr 12, 2021

conniey Apr 12, 2021

srnagar Apr 12, 2021

conniey Apr 12, 2021

Closing reactor connection and children. #19924

Closing reactor connection and children. #19924

Conversation

conniey commented Mar 17, 2021 • edited Loading

conniey commented Mar 17, 2021

azure-pipelines bot commented Mar 17, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

conniey commented Mar 17, 2021 •

edited

Loading