You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2026/10/22 17:12:51 UTC
(qpid-protonj2) branch 1.x updated: PROTON-2995 Notify closing links when their parent endpoint closes
This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch 1.x
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/1.x by this push:
new 70084549 PROTON-2995 Notify closing links when their parent endpoint closes
70084549 is described below
commit 700845499035edadef8962effc593437a1e6ea95
Author: Arnaud Cogoluègnes <51...@users.noreply.github.com>
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.
(cherry picked from commit 736a2341e88f02ce6ce2a5bd5fc43012d0c197a0)
---
.../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 8b2880ee..4c0ddec2 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: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org