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/test/java/org/opends/server/replication/protocol/SessionTest.java | 164 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 161 insertions(+), 3 deletions(-)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java
index f736e7b..79c0e25 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java
@@ -16,8 +16,10 @@
package org.opends.server.replication.protocol;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.Closeable;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
@@ -26,6 +28,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
@@ -138,6 +141,128 @@
}
}
+ /**
+ * A session with no thread writes on the publishing thread, as the session of a directory server
+ * does: a peer which reads nothing holds that write, and with it a close of the session, which
+ * waits for the lock of the writes. An abort ends both, and writes nothing more: the peer finds
+ * the message cut short and then the end of the stream, with no StopMsg.
+ */
+ @Test(timeOut = 60_000)
+ public void anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt() throws Exception
+ {
+ try (SessionPair pair = connectSessionPair())
+ {
+ final AtomicReference<Throwable> endOfTheWrite = new AtomicReference<>();
+ final Thread writer = start("writer", () -> {
+ try
+ {
+ pair.publisher.publish(newBlockingMsg());
+ }
+ catch (Throwable e)
+ {
+ endOfTheWrite.set(e);
+ }
+ });
+ pair.awaitBytesReachedThePeer();
+ final Thread closer = start("closer", pair.publisher::close);
+ awaitWaitingIn(closer, "close");
+
+ final Thread aborter = start("aborter",
+ () -> pair.publisher.abort(new IOException("the test gives the session up")));
+ aborter.join(SOCKET_TIMEOUT_MS);
+ writer.join(SOCKET_TIMEOUT_MS);
+ closer.join(SOCKET_TIMEOUT_MS);
+ assertThat(aborter.isAlive()).as("the abort waits for the write").isFalse();
+ assertThat(writer.isAlive()).as("the write still blocks after the abort").isFalse();
+ assertThat(endOfTheWrite.get()).as("how the write ended").isInstanceOf(IOException.class);
+ assertThat(closer.isAlive()).as("the close still waits for the write after the abort").isFalse();
+ assertThatThrownBy(pair.peer::receive)
+ .as("the peer read a message after the one the abort cut short")
+ .isInstanceOf(IOException.class);
+ }
+ }
+
+ /**
+ * The thread of a session can be the one held in a write, and any close joins that thread first:
+ * the abort ends the write before it joins it.
+ */
+ @Test(timeOut = 60_000)
+ public void anAbortEndsAWriteOfTheThreadOfTheSession() throws Exception
+ {
+ try (SessionPair pair = connectSessionPair())
+ {
+ pair.publisher.start();
+ pair.publisher.waitForStartup();
+ pair.publisher.publish(newBlockingMsg());
+ pair.awaitBytesReachedThePeer();
+
+ final Thread aborter = start("aborter",
+ () -> pair.publisher.abort(new IOException("the test gives the session up")));
+ aborter.join(SOCKET_TIMEOUT_MS);
+ assertThat(aborter.isAlive()).as("the abort waits for the thread of the session, held in its write")
+ .isFalse();
+ assertThat(pair.publisher.isAlive()).as("the thread of the session outlived the abort").isFalse();
+ assertThatThrownBy(pair.peer::receive)
+ .as("the peer read a message after the one the abort cut short")
+ .isInstanceOf(IOException.class);
+ }
+ }
+
+ /**
+ * An abort marks the close begun before it closes the sockets: a reader whose read the close of
+ * the sockets ends then finds the close begun, and does not take it for a failure of its peer -
+ * which the broker of a domain reports as a disconnection by its replication server.
+ */
+ @Test(timeOut = 60_000)
+ public void anAbortMarksTheCloseBegunBeforeItClosesTheSockets() throws Exception
+ {
+ try (SessionPair pair = connectSessionPair())
+ {
+ pair.peer.abort(new IOException("the test gives the session up"));
+ assertThat(pair.peerSocket.closeBegunWhenClosed())
+ .as("whether the close of the session was marked begun when its socket was closed")
+ .isTrue();
+ }
+ }
+
+ private static Thread start(final String name, final Runnable runnable)
+ {
+ final Thread thread = new Thread(runnable, SessionTest.class.getSimpleName() + " " + name);
+ thread.setDaemon(true);
+ thread.start();
+ return thread;
+ }
+
+ /** Waits for the thread to wait, for a lock or for a time, inside the method of a session. */
+ private static void awaitWaitingIn(final Thread thread, final String method) throws Exception
+ {
+ final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS);
+ while (!waitsIn(thread, method))
+ {
+ assertThat(System.nanoTime() < deadline)
+ .as(thread.getName() + " does not wait in Session." + method + "()")
+ .isTrue();
+ Thread.sleep(10);
+ }
+ }
+
+ private static boolean waitsIn(final Thread thread, final String method)
+ {
+ final Thread.State state = thread.getState();
+ if (state != Thread.State.WAITING && state != Thread.State.TIMED_WAITING)
+ {
+ return false;
+ }
+ for (final StackTraceElement frame : thread.getStackTrace())
+ {
+ if (frame.getClassName().equals(Session.class.getName()) && frame.getMethodName().equals(method))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
private static EntryMsg newBlockingMsg()
{
return new EntryMsg(SENDER_ID, PEER_ID, new byte[BLOCKING_MESSAGE_SIZE], 1);
@@ -160,7 +285,7 @@
{
final ReplSessionSecurity security = getReplSessionSecurity();
final ExecutorService executor = Executors.newSingleThreadExecutor();
- final Socket peerSocket = new Socket();
+ final ObservedSocket peerSocket = new ObservedSocket();
Socket publisherSocket = null;
Session publisher = null;
boolean connected = false;
@@ -180,6 +305,7 @@
publisher = security.createServerSession(publisherSocket, SOCKET_TIMEOUT_MS);
assertThat(publisher).as("could not create the session under test").isNotNull();
final Session peer = peerEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ peerSocket.observe(peer);
publisher.publish(new HeartbeatMsg());
assertThat(peer.receive()).isInstanceOf(HeartbeatMsg.class);
@@ -204,13 +330,45 @@
}
}
+ /**
+ * The socket of the peer, which records, when it is first closed, whether the close of the
+ * session it carries was marked begun by then.
+ */
+ private static final class ObservedSocket extends Socket
+ {
+ private volatile Session session;
+ private volatile Boolean closeBegunWhenClosed;
+
+ void observe(final Session session)
+ {
+ this.session = session;
+ }
+
+ /** Returns null until the socket of a session observed is closed. */
+ Boolean closeBegunWhenClosed()
+ {
+ return closeBegunWhenClosed;
+ }
+
+ @Override
+ public void close() throws IOException
+ {
+ final Session observed = session;
+ if (observed != null && closeBegunWhenClosed == null)
+ {
+ closeBegunWhenClosed = observed.closeInitiated();
+ }
+ super.close();
+ }
+ }
+
private static final class SessionPair implements Closeable
{
private final Session publisher;
private final Session peer;
- private final Socket peerSocket;
+ private final ObservedSocket peerSocket;
- private SessionPair(Session publisher, Session peer, Socket peerSocket)
+ private SessionPair(Session publisher, Session peer, ObservedSocket peerSocket)
{
this.publisher = publisher;
this.peer = peer;
--
Gitblit v1.10.0