external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -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;