This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/main by this push:
new 736a2341 PROTON-2995 Notify closing links when their parent endpoint
closes
736a2341 is described below
commit 736a2341e88f02ce6ce2a5bd5fc43012d0c197a0
Author: Arnaud Cogoluègnes <[email protected]>
AuthorDate: Thu Oct 22 16:02:54 2026 +0200
PROTON-2995 Notify closing links when their parent endpoint closes
A link closed or detached locally and waiting for the remote detach was
not notified when its parent session or connection was closed. As the
remote detach never arrives after the session end, closing the link from
the client blocked until the close timeout (60 seconds by default), for
example when the remote answers a detach with an end.
The parent endpoint closed handler is now also fired for links that are
locally closed or detached but not yet remotely closed or detached.
---
.../qpid/protonj2/client/impl/ReceiverTest.java | 61 +++++++++++
.../qpid/protonj2/client/impl/SenderTest.java | 60 ++++++++++
.../qpid/protonj2/engine/impl/ProtonLink.java | 5 +-
.../protonj2/engine/impl/ProtonReceiverTest.java | 122 +++++++++++++++++++++
.../protonj2/engine/impl/ProtonSenderTest.java | 122 +++++++++++++++++++++
5 files changed, 369 insertions(+), 1 deletion(-)
diff --git
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ReceiverTest.java
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ReceiverTest.java
index c59f828c..2c42043b 100644
---
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ReceiverTest.java
+++
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ReceiverTest.java
@@ -464,6 +464,67 @@ public class ReceiverTest extends ImperativeClientTestCase
{
}
}
+ @Test
+ public void
testCloseReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching() throws
Exception {
+
doTestCloseOrDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching(true,
false);
+ }
+
+ @Test
+ public void
testCloseReceiverCompletesWhenRemoteEndsSessionWithErrorInsteadOfDetaching()
throws Exception {
+
doTestCloseOrDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching(true,
true);
+ }
+
+ @Test
+ public void
testDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching() throws
Exception {
+
doTestCloseOrDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching(false,
false);
+ }
+
+ @Test
+ public void
testDetachReceiverCompletesWhenRemoteEndsSessionWithErrorInsteadOfDetaching()
throws Exception {
+
doTestCloseOrDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching(false,
true);
+ }
+
+ private void
doTestCloseOrDetachReceiverCompletesWhenRemoteEndsSessionInsteadOfDetaching(boolean
close, boolean error) throws Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
+ peer.expectFlow();
+ peer.expectDetach();
+ if (error) {
+
peer.remoteEnd().withErrorCondition(AmqpError.INTERNAL_ERROR.toString(),
"Cannot detach the link").queue();
+ } else {
+ peer.remoteEnd().queue();
+ }
+ peer.expectEnd();
+ peer.expectClose().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Receiver test started, peer listening on: {}",
remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort());
+ Session session = connection.openSession();
+ Receiver receiver = session.openReceiver("test-queue");
+ receiver.openFuture().get(10, TimeUnit.SECONDS);
+
+ // The remote detach never arrives once the session is ended, so
the close
+ // or detach must complete without waiting for the link close
timeout.
+ if (close) {
+ receiver.closeAsync().get(10, TimeUnit.SECONDS);
+ } else {
+ receiver.detachAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ connection.closeAsync().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
@Test
public void testReceiverDrainAllOutstanding() throws Exception {
try (ProtonTestServer peer = new ProtonTestServer()) {
diff --git
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
index f54e2054..7d6f7d36 100644
---
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
+++
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
@@ -524,6 +524,66 @@ public class SenderTest extends ImperativeClientTestCase {
}
}
+ @Test
+ public void
testCloseSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching() throws
Exception {
+
doTestCloseOrDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching(true,
false);
+ }
+
+ @Test
+ public void
testCloseSenderCompletesWhenRemoteEndsSessionWithErrorInsteadOfDetaching()
throws Exception {
+
doTestCloseOrDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching(true,
true);
+ }
+
+ @Test
+ public void
testDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching() throws
Exception {
+
doTestCloseOrDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching(false,
false);
+ }
+
+ @Test
+ public void
testDetachSenderCompletesWhenRemoteEndsSessionWithErrorInsteadOfDetaching()
throws Exception {
+
doTestCloseOrDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching(false,
true);
+ }
+
+ private void
doTestCloseOrDetachSenderCompletesWhenRemoteEndsSessionInsteadOfDetaching(boolean
close, boolean error) throws Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+ peer.expectDetach();
+ if (error) {
+
peer.remoteEnd().withErrorCondition(AmqpError.INTERNAL_ERROR.toString(),
"Cannot detach the link").queue();
+ } else {
+ peer.remoteEnd().queue();
+ }
+ peer.expectEnd();
+ peer.expectClose().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Sender test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort());
+ Session session = connection.openSession();
+ Sender sender = session.openSender("test-queue");
+ sender.openFuture().get(10, TimeUnit.SECONDS);
+
+ // The remote detach never arrives once the session is ended, so
the close
+ // or detach must complete without waiting for the link close
timeout.
+ if (close) {
+ sender.closeAsync().get(10, TimeUnit.SECONDS);
+ } else {
+ sender.detachAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ connection.closeAsync().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
@Test
public void testSendTimesOutWhenNoCreditIssued() throws Exception {
try (ProtonTestServer peer = new ProtonTestServer()) {
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/engine/impl/ProtonLink.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/engine/impl/ProtonLink.java
index 3cc7a8d2..846d5850 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/engine/impl/ProtonLink.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/engine/impl/ProtonLink.java
@@ -513,7 +513,10 @@ public abstract class ProtonLink<L extends Link<L>>
extends ProtonEndpoint<L> im
}
L fireParentEndpointClosed() {
- if (parentEndpointClosedEventHandler != null && isLocallyOpen()) {
+ // A link that was locally closed or detached but is still awaiting
the remote response
+ // must also be notified as that response will never arrive once the
parent is closed.
+ if (parentEndpointClosedEventHandler != null &&
+ (isLocallyOpen() || (isLocallyClosedOrDetached() &&
!isRemotelyClosedOrDetached()))) {
parentEndpointClosedEventHandler.handle(self());
}
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonReceiverTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonReceiverTest.java
index 4c930c5b..0c1a9698 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonReceiverTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonReceiverTest.java
@@ -294,6 +294,128 @@ public class ProtonReceiverTest extends
ProtonEngineTestSupport {
assertNull(failure);
}
+ @Test
+ public void testLocallyClosedReceiverReceivesParentSessionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach();
+ // The remote ends the session in response to the detach instead of
answering it
+ peer.remoteEnd().queue();
+ peer.expectEnd();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Receiver receiver = session.receiver("test");
+ receiver.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ receiver.open();
+ receiver.close();
+
+ assertTrue(session.isRemotelyClosed());
+ assertFalse(receiver.isRemotelyClosed());
+ assertFalse(parentClosed.get(), "Receiver should not have reported
parent session closed");
+
+ session.close();
+
+ assertTrue(parentClosed.get(), "Receiver awaiting remote detach should
have reported parent session closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
+ @Test
+ public void testLocallyClosedReceiverReceivesParentConnectionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach();
+ // The remote closes the connection in response to the detach instead
of answering it
+ peer.remoteClose().queue();
+ peer.expectClose();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Receiver receiver = session.receiver("test");
+ receiver.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ receiver.open();
+ receiver.close();
+
+ assertTrue(connection.isRemotelyClosed());
+ assertFalse(receiver.isRemotelyClosed());
+ assertFalse(parentClosed.get(), "Receiver should not have reported
parent connection closed");
+
+ connection.close();
+
+ assertTrue(parentClosed.get(), "Receiver awaiting remote detach should
have reported parent connection closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
+ @Test
+ public void testClosedReceiverDoesNotReceiveParentSessionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach().respond();
+ peer.expectEnd().respond();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Receiver receiver = session.receiver("test");
+ receiver.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ receiver.open();
+ receiver.close();
+
+ assertTrue(receiver.isRemotelyClosed());
+
+ session.close();
+
+ assertFalse(parentClosed.get(), "Receiver closed on both ends should
not have reported parent session closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
@Test
public void testEngineShutdownEventNeitherEndClosed() throws Exception {
doTestEngineShutdownEvent(false, false);
diff --git
a/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonSenderTest.java
b/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonSenderTest.java
index 08b0234f..3e30d7fb 100644
---
a/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonSenderTest.java
+++
b/protonj2/src/test/java/org/apache/qpid/protonj2/engine/impl/ProtonSenderTest.java
@@ -388,6 +388,128 @@ public class ProtonSenderTest extends
ProtonEngineTestSupport {
assertNull(failure);
}
+ @Test
+ public void testLocallyClosedSenderReceivesParentSessionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach();
+ // The remote ends the session in response to the detach instead of
answering it
+ peer.remoteEnd().queue();
+ peer.expectEnd();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Sender sender = session.sender("test");
+ sender.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ sender.open();
+ sender.close();
+
+ assertTrue(session.isRemotelyClosed());
+ assertFalse(sender.isRemotelyClosed());
+ assertFalse(parentClosed.get(), "Sender should not have reported
parent session closed");
+
+ session.close();
+
+ assertTrue(parentClosed.get(), "Sender awaiting remote detach should
have reported parent session closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
+ @Test
+ public void testLocallyClosedSenderReceivesParentConnectionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach();
+ // The remote closes the connection in response to the detach instead
of answering it
+ peer.remoteClose().queue();
+ peer.expectClose();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Sender sender = session.sender("test");
+ sender.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ sender.open();
+ sender.close();
+
+ assertTrue(connection.isRemotelyClosed());
+ assertFalse(sender.isRemotelyClosed());
+ assertFalse(parentClosed.get(), "Sender should not have reported
parent connection closed");
+
+ connection.close();
+
+ assertTrue(parentClosed.get(), "Sender awaiting remote detach should
have reported parent connection closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
+ @Test
+ public void testClosedSenderDoesNotReceiveParentSessionClosedEvent()
throws Exception {
+ Engine engine = EngineFactory.PROTON.createNonSaslEngine();
+ engine.errorHandler(result -> failure = result.failureCause());
+ ProtonTestConnector peer = createTestPeer(engine);
+
+ final AtomicBoolean parentClosed = new AtomicBoolean();
+
+ peer.expectAMQPHeader().respondWithAMQPHeader();
+ peer.expectOpen().respond().withContainerId("driver");
+ peer.expectBegin().respond();
+ peer.expectAttach().respond();
+ peer.expectDetach().respond();
+ peer.expectEnd().respond();
+
+ Connection connection = engine.start();
+
+ connection.open();
+ Session session = connection.session();
+ session.open();
+
+ Sender sender = session.sender("test");
+ sender.parentEndpointClosedHandler(result -> parentClosed.set(true));
+
+ sender.open();
+ sender.close();
+
+ assertTrue(sender.isRemotelyClosed());
+
+ session.close();
+
+ assertFalse(parentClosed.get(), "Sender closed on both ends should not
have reported parent session closed");
+
+ peer.waitForScriptToComplete();
+
+ assertNull(failure);
+ }
+
@Test
public void testEngineShutdownEventNeitherEndClosed() throws Exception {
doTestEngineShutdownEvent(false, false);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
