From 77ad03e14cc1bbc5471f332ba73430d4c472b366 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 20 Oct 2026 06:45:44 +0000
Subject: [PATCH] [#1202] [#1209] [#1222] Give the replication server a bounded time to give credit and to read once the broker is being stopped, and write nothing after a message given up (#1210)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java | 51 ++++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 46 insertions(+), 5 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
index f92154f..a7823e3 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
@@ -194,18 +194,35 @@
@Override
public void close()
{
+ close(false);
+ }
+
+ /**
+ * Closes the session, as {@link #close()} or as {@link #abort(IOException)} does.
+ *
+ * @param abort
+ * whether the sockets are to be closed at once, before anything waits for the
+ * publisher thread or for {@code publishLock} - even when a close had begun already
+ */
+ private void close(final boolean abort)
+ {
Throwable localSessionError;
+ final boolean closing;
synchronized (stateLock)
{
- if (closeInitiated)
- {
- return;
- }
-
+ closing = closeInitiated;
localSessionError = sessionError;
closeInitiated = true;
}
+ if (abort)
+ {
+ StaticUtils.close(plainSocket, secureSocket);
+ }
+ if (closing)
+ {
+ return;
+ }
try {
interrupt();
@@ -319,6 +336,30 @@
StaticUtils.close(plainSocket, secureSocket);
}
+ /**
+ * Closes this session at once, as a session which failed with the error provided: nothing more
+ * is written to it - neither what its publisher thread left queued nor the StopMsg - and a write
+ * blocked in its socket, because the peer reads nothing, fails rather than holds the close.
+ * <p>
+ * {@link #close()} of a session with no error takes {@code publishLock} before it closes the
+ * sockets, and a write blocked in the socket holds that lock for as long as the peer does not
+ * read, which TCP does not bound; and any close first joins the publisher thread, which may be
+ * the writer. So the error is recorded first, and the close is marked begun, so that a reader
+ * whose read the close of the sockets ends finds it begun - see {@link #closeInitiated()}. Then
+ * the sockets are closed, before the join: that ends the write, whichever thread makes it, and
+ * releases a close which had begun before, with no error, and waits for the lock - the call here
+ * then returns at once. The close itself takes the branch of a session which failed, and writes
+ * nothing.
+ *
+ * @param cause
+ * why the session is given up, which it then reports as its error
+ */
+ public void abort(final IOException cause)
+ {
+ setSessionError(cause);
+ close(true);
+ }
+
/**
--
Gitblit v1.10.0