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 +
opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java | 164 ++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 6
opendj-server-legacy/src/main/java/org/opends/server/replication/service/HeartbeatMonitor.java | 24
opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml | 17
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java | 275 ++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 6
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java | 1727 +++++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java | 6
9 files changed, 2,239 insertions(+), 37 deletions(-)
diff --git a/opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml b/opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml
index 0201305..59ef52c 100644
--- a/opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml
+++ b/opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml
@@ -14,6 +14,7 @@
Copyright 2007-2008 Sun Microsystems, Inc.
Portions Copyright 2011-2015 ForgeRock AS.
+ Portions Copyright 2026 3A Systems, LLC.
! -->
<adm:managed-object name="replication-synchronization-provider"
plural-name="replication-synchronization-providers"
@@ -97,8 +98,22 @@
<adm:property name="connection-timeout" advanced="true">
<adm:synopsis>
Specifies the timeout used when connecting to peers and when performing
- SSL negotiation.
+ SSL negotiation, and how long a stopping replication domain waits for
+ window credit before it gives up the changes it still has to send.
</adm:synopsis>
+ <adm:description>
+ A replication domain is stopped when it is disabled, when the server
+ shuts down, and before an import, a restore or a restart of its session.
+ A stopping domain gives its replication server this long to give the
+ window credit for the changes it still has to send, and for the message
+ which announces it offline: raising the value lengthens a stop against a
+ replication server which gives no credit, lowering it shortens the time
+ before those changes are given up. A replication server which does not
+ read what the domain has already written to it gets as long again, after
+ which the domain closes the session without waiting any longer. A value
+ of 0 sets no limit: a connection, an SSL negotiation and a stopping
+ replication domain then wait for as long as it takes.
+ </adm:description>
<adm:default-behavior>
<adm:defined>
<adm:value>5 seconds</adm:value>
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index 5cd3bac..9215e9f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -2574,8 +2574,10 @@
*/
logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
+ " could not announce itself offline: the message was not published - a change which"
- + " is still in flight holds it back, or the broker had no session to write it to,"
- + " and " + pendingChanges.size() + " change(s) are pending");
+ + " is still in flight holds it back, the broker had no session to write it to, or it"
+ + " got no window credit for it, or for a change before it, or the replication server"
+ + " did not read what was written to it, in the time the stop gives the replication server"
+ + " - and " + pendingChanges.size() + " change(s) are pending");
}
}
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);
+ }
+
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java
index 21e5aee..361af54 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java
@@ -146,8 +146,10 @@
* period; the peers it is queued for, if they are recorded in that window, are lost too, with
* the opposite effect - the first forward ends its wait, as for a message no peer was recorded
* for. That window is the one publish the broker refuses: at once on a connection error or a
- * pending recovery, the broker's retry loop up to the reconnect when it has no session. The
- * wait it can cost is bounded by a grace period which is already running.
+ * pending recovery, the broker's retry loop up to the reconnect when it has no session, and the
+ * wait of a stopping broker for window credit, or for its replication server to read, up to the
+ * deadline of the stop. The wait it can cost is bounded by a grace period which is already
+ * running.
*
* @param baseDN
* the domain for which the message was announced
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/HeartbeatMonitor.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/HeartbeatMonitor.java
index f889968..affe18c 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/HeartbeatMonitor.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/HeartbeatMonitor.java
@@ -13,11 +13,15 @@
*
* Copyright 2007-2009 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.service;
import static org.opends.messages.ReplicationMessages.*;
+
+import java.io.IOException;
+
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.protocol.Session;
@@ -107,6 +111,7 @@
heartbeatInterval);
}
+ boolean heartbeatsStopped = false;
try
{
boolean gotOneFailure = false;
@@ -124,6 +129,7 @@
// Exit monitor and close session.
shutdown = true;
+ heartbeatsStopped = true;
break;
}
else
@@ -161,7 +167,23 @@
{
logger.trace("Heartbeat monitor is exiting");
}
- session.close();
+ if (heartbeatsStopped)
+ {
+ /*
+ * A replication server which is gone without a reset acknowledges nothing, and the writes to
+ * it fill the socket buffers: a write then blocks in the socket, holding the lock a close of
+ * a session with no error waits for, until TCP gives up retransmitting. Nothing else ends
+ * that write, nor the read of the listener, which neither data nor a reset ends: the
+ * session is closed as a failed one, with no StopMsg the replication server would not read.
+ */
+ session.abort(new IOException("No heartbeat came from replication server RS("
+ + replicationServerID + ") at " + session.getReadableRemoteAddress() + " for twice "
+ + heartbeatInterval + " ms"));
+ }
+ else
+ {
+ session.close();
+ }
}
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
index ae97bd3..fc2fb21 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
@@ -51,6 +51,7 @@
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.server.ReplicationDomainCfg;
import org.forgerock.util.Utils;
+import org.opends.server.api.DirectoryThread;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.DSInfo;
@@ -166,8 +167,91 @@
}
}
+ /**
+ * A stop of the broker, from the moment {@link #stop()} begins until {@link #start()} clears it.
+ * <p>
+ * What a publish gives up for the stop is marked on the stop, and not on the broker: a publish
+ * which found the stop past its deadline may mark it only after start() cleared it, and then the
+ * mark lands on a stop which is over, rather than on the broker started again.
+ */
+ private static final class StopInProgress
+ {
+ /**
+ * The {@link System#nanoTime()} until which a publish waits for credit. The replication server
+ * gets as long to give the credit as it gets to accept a connection: one which is only slow
+ * takes the changes and the replica offline message as before, and only one which takes
+ * nothing for that long costs them.
+ */
+ private final long creditDeadline;
+ /**
+ * The {@link System#nanoTime()} past which a stop still held is ended by the close of the
+ * session - see {@link ReplicationBroker#endWhatStillHoldsTheStop(StopInProgress)}. It comes as
+ * long after {@link #creditDeadline} as that comes after the start of the stop: a replication
+ * server which reads slowly gets as long to read what was written as it got to give credit.
+ */
+ private final long writeDeadline;
+ /**
+ * Whether the stop has a deadline: a connection timeout of 0 lets a connection wait for as
+ * long as it takes, and the stop as well, as it did before it had any deadline.
+ */
+ private final boolean bounded;
+ /**
+ * Set when a publish gave its message up because of this stop, or when the stop closed its
+ * session: from then on publish() writes nothing, and the broker opens no session again, until
+ * start() clears the stop. The change time heartbeats and the window and status messages
+ * written straight to the session are not held back. A message published after one given up
+ * would reach the replication server ahead of an older one - the replica offline message,
+ * which carries the newest CSN of the replica, would tell it that nothing older is to come.
+ */
+ private volatile boolean gaveUp;
+ /**
+ * Set once stop() is over, by {@link #end()}. Guarded by this object, which the stop holds
+ * while it closes the session, so that the close never comes after the end of the stop.
+ */
+ private boolean ended;
+ /** The connection timeout the deadlines were set from, in milliseconds. */
+ private final long connectionTimeoutMS;
+
+ private StopInProgress(long connectionTimeoutMS)
+ {
+ final long timeout = TimeUnit.MILLISECONDS.toNanos(connectionTimeoutMS);
+ this.creditDeadline = System.nanoTime() + timeout;
+ this.writeDeadline = creditDeadline + timeout;
+ this.bounded = connectionTimeoutMS > 0;
+ this.connectionTimeoutMS = connectionTimeoutMS;
+ }
+
+ private boolean isPastTheCreditDeadline()
+ {
+ return bounded && System.nanoTime() - creditDeadline >= 0;
+ }
+
+ /** Ends the stop, once a close of its session in progress is done; none comes after it. */
+ private synchronized void end()
+ {
+ ended = true;
+ notifyAll();
+ }
+ }
+
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
private volatile boolean shutdown;
+ /**
+ * The stop in progress: set by {@link #stop()} before anything else it does, and cleared to
+ * {@code null} by {@link #start()}. While it is set, a publish which finds no window credit by
+ * its deadline gives its message up instead of waiting on.
+ * <p>
+ * stop() publishes the replica offline message before it sets {@link #shutdown}, the flag which
+ * ends the wait for credit, and that message needs credit too, and the lock of the pending
+ * changes a local change waits for credit under. Credit comes only as the replication server
+ * takes the updates of the session, while its heartbeats keep the session up: without a bound a
+ * replication server which takes no update holds the stop - and the disable, the restart or the
+ * shutdown of the domain behind it - for as long as it does. A replication server which reads
+ * nothing holds it as well, through a write which blocks in the socket once its buffers are full,
+ * with the locks the write is made under: that one only the close of the socket ends, and the
+ * stop closes the session past a second deadline.
+ */
+ private volatile StopInProgress stopping;
private final Object startStopLock = new Object();
private volatile ReplicationDomainCfg config;
/** String reported under CSN=monitor when there is no connected RS. */
@@ -287,6 +371,7 @@
return;
}
shutdown = false;
+ stopping = null;
this.rcvWindow = getMaxRcvWindow();
connectAsDataServer();
}
@@ -2220,15 +2305,27 @@
break;
}
- try
+ /*
+ * Once the stop gave a message up, nothing more is published, and a session opened while the
+ * stop runs would carry only its change time heartbeats - which tell the replication server
+ * that nothing older is to come from this replica, while it still has the changes given up -
+ * and hold the stop for as long as the connection takes: the stop waits for this lock. The
+ * loop waits for the end of the stop instead - the listener gets here when the stop closes
+ * the session it reads. A connection already under way when the stop gives up still opens
+ * its session, which publish() then writes nothing to, and which the stop closes as it ends.
+ */
+ if (!gaveUpForTheStop())
{
- connectAsDataServer();
- rs = connectedRS.get();
- }
- catch (Exception e)
- {
- logger.error(NOTE_EXCEPTION_RESTARTING_SESSION,
- getBaseDN(), e.getLocalizedMessage() + " " + stackTraceToSingleLineString(e));
+ try
+ {
+ connectAsDataServer();
+ rs = connectedRS.get();
+ }
+ catch (Exception e)
+ {
+ logger.error(NOTE_EXCEPTION_RESTARTING_SESSION,
+ getBaseDN(), e.getLocalizedMessage() + " " + stackTraceToSingleLineString(e));
+ }
}
if (rs.isConnected() || !infiniteTry)
@@ -2319,7 +2416,7 @@
{
boolean done = false;
- while (!done && !shutdown)
+ while (!done && !shutdown && !gaveUpForTheStop())
{
if (connectionError)
{
@@ -2400,6 +2497,7 @@
credit = true;
}
+ final StopInProgress stop = stopping;
if (credit)
{
synchronized (connectPhaseLock)
@@ -2410,6 +2508,18 @@
Check the session. If it has changed, some disconnection or
reconnection happened and we need to restart from scratch.
*/
+ if (gaveUpForTheStop())
+ {
+ /*
+ * The stop gave up since the loop head, while this publish waited for this lock - for
+ * a connection, say, which held it past the deadline of the watchdog of the stop, and
+ * left it no session to close: the loop head decides, rather than this publish writes
+ * to the session that connection opened. The watchdog marks the give-up before it
+ * reads the session, and a connection sets its session before it releases this lock:
+ * either the watchdog closes the session read here, or this read finds the mark.
+ */
+ continue;
+ }
final Session session = connectedRS.get().session;
if (session != null && session == currentSession)
{
@@ -2418,6 +2528,25 @@
}
}
}
+ else if (stop != null && stop.isPastTheCreditDeadline())
+ {
+ if (stop != stopping)
+ {
+ // start() ran since the read: the loop head decides, as before a stop had a deadline.
+ continue;
+ }
+ /*
+ * Only credit ends this wait on a session the replication server keeps up, and stop()
+ * sets the flag of this loop only after the replica offline message, which may need the
+ * lock this publish holds. What is given up here is a change which the next session
+ * republishes - from the historical information of its entry, or a local delete, one the
+ * recovery publishes included, from the deletes this domain keeps aside in memory, which
+ * a restart of the server loses - or the replica offline message, which is then reported
+ * as not sent.
+ */
+ stop.gaveUp = true;
+ return false;
+ }
if (!credit && currentWindowSemaphore.availablePermits() == 0)
{
synchronized (connectPhaseLock)
@@ -2427,9 +2556,10 @@
Send a WindowProbeMsg message to wake up the receiver in case the
window update message was lost somehow...
then loop to check again if connection was closed.
+ Not once the stop gave up, which it may have done while this waited for the lock.
*/
Session session = connectedRS.get().session;
- if (session != null)
+ if (session != null && !gaveUpForTheStop())
{
session.publish(new WindowProbeMsg());
}
@@ -2449,7 +2579,8 @@
}
// The receive threads should handle reconnection or
- // mark this broker in error. Just retry.
+ // mark this broker in error. Just retry - unless the stop closed the session to end this
+ // write: it marked the give-up first, and the loop ends on it.
synchronized (connectPhaseLock)
{
try
@@ -2477,13 +2608,20 @@
}
}
/*
- * The loop also ends when the broker is stopped, and then nothing was written to any
- * session: a caller told the message was published would report as sent a message which
- * never reached the wire.
+ * The loop also ends when the broker is stopped, or gave a message up for the stop, and then
+ * nothing was written to any session: a caller told the message was published would report as
+ * sent a message which never reached the wire.
*/
return done;
}
+ /** Tells whether a publish gave its message up for the stop in progress, if there is one. */
+ private boolean gaveUpForTheStop()
+ {
+ final StopInProgress stop = stopping;
+ return stop != null && stop.gaveUp;
+ }
+
/**
* Receive a message.
* This method is not thread-safe and should either always be
@@ -2765,21 +2903,114 @@
{
debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
}
+ final StopInProgress stop;
+ final boolean newStop;
synchronized (startStopLock) {
if (shutdown) {
return;
}
+ // A stop which overlaps one in progress keeps its deadlines and what it gave up.
+ newStop = stopping == null;
+ if (newStop)
+ {
+ stopping = new StopInProgress(MultimasterReplication.getConnectionTimeoutMS());
+ }
+ stop = stopping;
}
- domain.publishReplicaOfflineMsg();
-
- synchronized (startStopLock)
+ try
{
- shutdown = true;
- setConnectedRS(ConnectedRS.stopped());
- stopRSHeartBeatMonitoring();
- stopChangeTimeHeartBeatPublishing();
- deregisterReplicationMonitor();
+ // In the try: a watchdog which cannot be started must not leave the broker stopping either.
+ if (newStop && stop.bounded)
+ {
+ new DirectoryThread(() -> endWhatStillHoldsTheStop(stop),
+ "Replica DS(" + getServerId() + ") watchdog of the stop for domain \"" + getBaseDN() + "\"")
+ .start();
+ }
+ domain.publishReplicaOfflineMsg();
+ }
+ finally
+ {
+ // A broker left running as a stopping one would give up every publish which waits for credit.
+ synchronized (startStopLock)
+ {
+ try
+ {
+ shutdown = true;
+ setConnectedRS(ConnectedRS.stopped());
+ stopRSHeartBeatMonitoring();
+ stopChangeTimeHeartBeatPublishing();
+ deregisterReplicationMonitor();
+ }
+ finally
+ {
+ // Under the lock, so that no start() opens a session which this stop could still close.
+ stop.end();
+ }
+ }
+ }
+ }
+
+ /**
+ * Waits for the stop provided to end, and closes the session if the stop is still held once it is
+ * past its {@link StopInProgress#writeDeadline}.
+ * <p>
+ * A publish which waits for credit gives up by itself once the stop is past its credit deadline,
+ * and the stop then ends the session in order. What holds the stop long after that is a write the
+ * replication server does not read - of a change which got its credit, of a window probe, of the
+ * StopMsg which ends the session - or a lock which such a write holds: the lock of the pending
+ * changes, which the replica offline message needs, the lock of the connection phase, or the lock
+ * of the writes of the session, which a close of a session with no error takes. A socket write
+ * has no timeout, and returns only once the peer reads or the socket is closed: the session is
+ * closed here as a failed one, without that lock - see {@link Session#abort(IOException)}. The
+ * give-up is marked first, so that a publish whose write then fails returns rather than retries,
+ * and no session is opened again for the stop.
+ * <p>
+ * The close is made under the monitor of the stop, which {@link StopInProgress#end()} takes, and
+ * stop() ends before start() can run again: the session closed here is never one of a broker
+ * started again.
+ */
+ private void endWhatStillHoldsTheStop(final StopInProgress stop)
+ {
+ synchronized (stop)
+ {
+ long left;
+ while (!stop.ended && (left = stop.writeDeadline - System.nanoTime()) > 0)
+ {
+ try
+ {
+ TimeUnit.NANOSECONDS.timedWait(stop, left);
+ }
+ catch (InterruptedException e)
+ {
+ // Only a shutdown of the server interrupts this thread: the stop is left to itself.
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ if (stop.ended)
+ {
+ return;
+ }
+
+ stop.gaveUp = true;
+ /*
+ * Read after the mark: a connection under way leaves no session to close here, and publish()
+ * writes nothing to the one it then opens, as it reads the mark again under the lock that
+ * connection holds. The stop closes that session as it ends.
+ */
+ final ConnectedRS rs = connectedRS.get();
+ if (rs.session != null)
+ {
+ logger.warn(LocalizableMessage.raw("The stop of the replication domain %s of directory "
+ + "server %d still waits for its replication server RS(%d) at %s after %d ms, twice "
+ + "the connection timeout: its session is closed as a failed one, and nothing more is "
+ + "written to it",
+ getBaseDN(), getServerId(), rs.getServerId(), rs.replicationServer,
+ 2 * stop.connectionTimeoutMS));
+ rs.session.abort(new IOException("The stop of the domain gave the replication server "
+ + 2 * stop.connectionTimeoutMS + " ms to take what was written to the session"));
+ }
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 3dbca34..f72a4fe 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -3991,7 +3991,11 @@
* @return {@code true} if the message was written to the session of the
* Replication Service, {@code false} if the broker could not write
* it: it has no usable session, the changes which come before this
- * one still have to be republished, or it was stopped in between.
+ * one still have to be republished, or it was stopped in between -
+ * or is being stopped, and the window of its session gave it no
+ * credit for this message, or for one before it, or the replication
+ * server did not read what was written to it, in the time the stop
+ * gives the replication server.
*/
public boolean publish(UpdateMsg msg)
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java
new file mode 100644
index 0000000..4de4616
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java
@@ -0,0 +1,1727 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.plugin;
+
+import static java.util.concurrent.TimeUnit.*;
+
+import static org.forgerock.opendj.ldap.ModificationType.*;
+import static org.opends.messages.ReplicationMessages.WARN_REPLICATION_SERVER_BADLY_DISCONNECTED;
+import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.protocols.internal.InternalClientConnection.*;
+import static org.opends.server.protocols.internal.Requests.*;
+import static org.testng.Assert.*;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ResultCode;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.common.DSInfo;
+import org.opends.server.replication.common.RSInfo;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.DoneMsg;
+import org.opends.server.replication.protocol.EntryMsg;
+import org.opends.server.replication.protocol.HeartbeatMsg;
+import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ReplServerStartMsg;
+import org.opends.server.replication.protocol.ReplSessionSecurity;
+import org.opends.server.replication.protocol.ReplicaOfflineMsg;
+import org.opends.server.replication.protocol.ReplicationMsg;
+import org.opends.server.replication.protocol.ServerStartMsg;
+import org.opends.server.replication.protocol.Session;
+import org.opends.server.replication.protocol.StartSessionMsg;
+import org.opends.server.replication.protocol.StopMsg;
+import org.opends.server.replication.protocol.TopologyMsg;
+import org.opends.server.replication.protocol.WindowMsg;
+import org.opends.server.replication.protocol.WindowProbeMsg;
+import org.opends.server.replication.service.DSRSShutdownSync;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.util.TestTimer;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Tests that the broker of a domain can be stopped while its replication server keeps the session
+ * up and takes no update from it, and that the stop gives up no more than it has to.
+ * <p>
+ * The replication server returns window credit as its reader takes the updates of the session, and
+ * the same reader answers a window probe, while its heartbeats come from a thread of their own: a
+ * reader held up leaves the window closed on a session the domain sees alive. A publish then waits
+ * for credit with nothing but the stop of the broker to end it, and stop() publishes the replica
+ * offline message - which needs credit as well, and the lock of the pending changes a local change
+ * is pushed under - before it sets the flag which ends that wait.
+ * <p>
+ * A replication server which reads nothing of the session holds the stop as well, once the socket
+ * buffers are full: a write then blocks in the socket, with the locks it is made under, and only
+ * the close of the socket ends it. A replication server which is gone altogether, heartbeats
+ * included, holds that write with no stop at all: the heartbeat monitor has to close the session
+ * the same way.
+ * <p>
+ * The stop gives the replication server the connection timeout of the replication provider to give
+ * credit, and as long again to read what was written. The suite sets it to 30 seconds: the cases run
+ * with {@link #SHORT_CONNECTION_TIMEOUT_MS}, and those which need another one set it for their own
+ * run.
+ */
+@SuppressWarnings("javadoc")
+public class ReplicationBrokerStopTest extends ReplicationTestCase
+{
+ private static final int DS_ID = 1;
+ private static final int OTHER_DS_ID = 3;
+ private static final DN PROVIDER_DN =
+ DN.valueOf("cn=Multimaster Synchronization,cn=Synchronization Providers,cn=config");
+ /** The connection timeout the cases run with, unless they set their own. */
+ private static final long SHORT_CONNECTION_TIMEOUT_MS = 2000;
+ /**
+ * A connection timeout longer than the shipped default of 5 seconds, and far longer than what a
+ * replication server which is only slow needs in these cases to give credit.
+ */
+ private static final long LONG_CONNECTION_TIMEOUT_MS = 6000;
+ /** The window of a replication server whose cases need no wait for credit: the default one. */
+ private static final int LARGE_WINDOW = 100000;
+ /**
+ * The size of the messages a case writes until the socket buffers are full, in bytes or in
+ * characters: a few dozen of them fill the buffers a loopback connection gets on its own.
+ */
+ private static final int LARGE_MESSAGE_SIZE = 256 * 1024;
+ /** How many of those messages a case writes at most before it reports that no write blocks. */
+ private static final int MAX_LARGE_MESSAGES = 400;
+ /**
+ * The heartbeat interval of the case of a replication server which is gone: long enough for its
+ * writes to fill the socket buffers before the heartbeat monitor gives up on the session.
+ */
+ private static final long HEARTBEAT_INTERVAL_MS = 4000;
+ /**
+ * The replication servers which answer no connection in the case of a connection under way: each
+ * costs the connection the connection timeout, and together they hold it well past twice that, the
+ * deadline of the watchdog of a stop.
+ */
+ private static final int SILENT_SERVERS = 4;
+ /** A value which makes a change of the description one of those messages. */
+ private static final String LARGE_VALUE = "x".repeat(LARGE_MESSAGE_SIZE);
+
+ private final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ private long connectionTimeoutOfTheSuiteMS;
+
+ @BeforeClass
+ public void shortenTheConnectionTimeout()
+ {
+ connectionTimeoutOfTheSuiteMS = MultimasterReplication.getConnectionTimeoutMS();
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void restoreTheConnectionTimeout()
+ {
+ setConnectionTimeout(connectionTimeoutOfTheSuiteMS);
+ }
+
+ /**
+ * With the only credit of the window spent, the replica offline message stop() publishes finds
+ * none: the disable has to end all the same, and the message is given up rather than written -
+ * once the replication server had the connection timeout, as it is configured, to give credit.
+ * The announcement of the message is withdrawn, and the session is ended in order: a wait for
+ * credit, which gives up by itself, is no reason to close it as a failed one.
+ */
+ @Test(timeOut = 120_000)
+ public void aDisableEndsWhenTheReplicationServerGivesNoCredit() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+
+ // The only credit of the window goes to a change of our own.
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final long beforeTheDisable = System.nanoTime();
+ disabling = startThread("disable", domain::disable);
+ disabling.join(stopBoundMS());
+ final long disableMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable);
+ assertFalse(disabling.isAlive(),
+ "disable() waits for window credit a replication server which takes no update does not give");
+ assertTrue(disableMS >= LONG_CONNECTION_TIMEOUT_MS, "disable() gave the replication server " + disableMS
+ + " ms to give credit, not the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 0, "the replica offline message was written with no credit");
+ assertEquals(rs.count(StopMsg.class), 1, "the session was closed as a failed one, not ended in order");
+
+ /* The grace period of the announcement has run out by now, so whether the shutdown of a
+ * collocated replication server would wait for it says nothing: what is left announced does.
+ * An announcement older than the change is one of an earlier stop of the replica. */
+ final CSN announced = offlineAnnouncementOfTheReplica();
+ assertTrue(announced == null || !announced.isNewerThan(rs.receivedUpdateCSNs().get(0)),
+ "the replica offline message given up is still announced: " + announced);
+ }
+ finally
+ {
+ cleanUp(rs, disabling);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * With no connection timeout, a connection waits for its replication server for as long as it
+ * takes, and so does a stop for credit: the replica offline message waits past its first wait for
+ * credit and past the deadline a stop would have with the shipped default of 5 seconds, and it
+ * goes out once the credit comes.
+ */
+ @Test(timeOut = 120_000)
+ public void withNoConnectionTimeoutAStopWaitsForTheCredit() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ setConnectionTimeout(0);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final long beforeTheDisable = System.nanoTime();
+ disabling = startThread("disable", domain::disable);
+ // A probe follows each wait for credit which found none, unless the publish gave up after it.
+ rs.waitForWindowProbes(2);
+ Thread.sleep(Math.max(0,
+ LONG_CONNECTION_TIMEOUT_MS + 1000 - NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable)));
+ assertTrue(disabling.isAlive(), "with no connection timeout, the stop gave the replica offline message up");
+
+ rs.giveCredit(1);
+ disabling.join(stopBoundMS());
+ assertFalse(disabling.isAlive(), "disable() did not end once the replication server gave credit");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
+ }
+ finally
+ {
+ cleanUp(rs, disabling);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * A local change which waits for credit holds the lock of the pending changes, which stop() needs
+ * to publish the replica offline message: the disable has to end the wait rather than wait for
+ * it, and the change ends with it.
+ */
+ @Test(timeOut = 120_000)
+ public void aDisableEndsWhileALocalChangeWaitsForCredit() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread writer = null;
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final AtomicReference<ResultCode> second = new AtomicReference<>();
+ writer = startThread("writer", () -> second.set(replaceDescription("second")));
+ // A probe is sent only by a publish which waited for credit and found the window closed.
+ rs.waitForWindowProbes(1);
+
+ disabling = startThread("disable", domain::disable);
+ disabling.join(stopBoundMS());
+ assertFalse(disabling.isAlive(),
+ "disable() waits for a local change which waits for window credit under the lock it needs");
+ writer.join(stopBoundMS());
+ assertFalse(writer.isAlive(), "the local change still waits for credit on a stopped broker");
+ assertEquals(second.get(), ResultCode.SUCCESS);
+ }
+ finally
+ {
+ cleanUp(rs, disabling, writer);
+ }
+ }
+
+ /**
+ * A replication server which is only slow gives the credit before the stop gives up on it: the
+ * change which waited for credit under the lock goes out, and the replica offline message after
+ * it, as they did before a stop could give anything up.
+ */
+ @Test(timeOut = 120_000)
+ public void aStopPublishesWhatTheReplicationServerTakesInTime() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ // The credit comes about a second into the stop: well within the time the stop gives.
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread writer = null;
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final AtomicReference<ResultCode> second = new AtomicReference<>();
+ writer = startThread("writer", () -> second.set(replaceDescription("second")));
+ rs.waitForWindowProbes(1);
+ disabling = startThread("disable", domain::disable);
+ waitForTheOfflineMessageToWaitForTheLock(disabling);
+
+ // The change keeps waiting for credit well into the stop: two more waits of it.
+ rs.waitForWindowProbes(rs.count(WindowProbeMsg.class) + 2);
+ rs.giveCredit(2);
+
+ disabling.join(stopBoundMS());
+ assertFalse(disabling.isAlive(), "disable() did not end once the replication server gave credit");
+ writer.join(stopBoundMS());
+ assertEquals(second.get(), ResultCode.SUCCESS);
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(LDAPUpdateMsg.class), 2, "the change which waited for credit was given up");
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
+ final List<Class<?>> updates = rs.receivedUpdates();
+ assertEquals(updates.get(updates.size() - 1), ReplicaOfflineMsg.class,
+ "the replica offline message has to come after the changes before it: " + updates);
+ }
+ finally
+ {
+ cleanUp(rs, disabling, writer);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * Once the stop gave a change up, the broker writes nothing more: the replica offline message
+ * carries the newest CSN of the replica, and written after a change it gave up it tells the
+ * replication server that the replica has nothing more to send up to that CSN. A broker started
+ * again finds the replica ahead of its replication server, and the recovery of its session
+ * publishes the change given up; then it writes as before.
+ */
+ @Test(timeOut = 120_000)
+ public void nothingIsWrittenAfterAChangeTheStopGaveUp() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ // The stop waits for the lock until the change gives up: long enough for the case to see it.
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread writer = null;
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final AtomicReference<ResultCode> second = new AtomicReference<>();
+ writer = startThread("writer", () -> second.set(replaceDescription("second")));
+ rs.waitForWindowProbes(1);
+ disabling = startThread("disable", domain::disable);
+ waitForTheOfflineMessageToWaitForTheLock(disabling);
+
+ /* The change is given up, and credit comes right after it: a stop which went on writing would
+ * still wait for it, while one which writes nothing may have ended the session already. */
+ writer.join(stopBoundMS());
+ assertFalse(writer.isAlive(), "the local change still waits for credit on a stopped broker");
+ rs.giveCreditIfTheSessionIsUp(1);
+
+ disabling.join(stopBoundMS());
+ assertFalse(disabling.isAlive(), "disable() waits for window credit");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(LDAPUpdateMsg.class), 1);
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
+ "the replica offline message was written after a change the stop gave up");
+
+ /* That is over once the broker is started again. The replication server has only the first
+ * change, and the domain the second as well: the recovery of the new session republishes it,
+ * with the credit of the window of that session. */
+ domain.enable();
+ waitForConnection(domain);
+ rs.waitForUpdates(2);
+ final List<CSN> csns = rs.receivedUpdateCSNs();
+ assertTrue(csns.get(1).isNewerThan(csns.get(0)) && csns.get(1).getServerId() == DS_ID,
+ "the change the stop gave up was not republished: " + csns);
+ waitForTheRecoveryToEnd();
+
+ rs.giveCredit(1);
+ assertEquals(replaceDescription("third"), ResultCode.SUCCESS);
+ rs.waitForUpdates(3);
+ }
+ finally
+ {
+ cleanUp(rs, disabling, writer);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * A restart of the session stops the broker as a disable does, and the change the stop gives up
+ * is republished by the recovery of the next session. Unlike enable(), which loads the state of
+ * the domain again and repairs it from the historical information of the entries, a restart
+ * keeps the state in memory: the change is known to be missing only if that state took it
+ * although the broker did not write it.
+ */
+ @Test(timeOut = 120_000)
+ public void aSessionRestartRepublishesTheChangeTheStopGaveUp() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread writer = null;
+ Thread restarting = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final AtomicReference<ResultCode> second = new AtomicReference<>();
+ writer = startThread("writer", () -> second.set(replaceDescription("second")));
+ rs.waitForWindowProbes(1);
+ restarting = startThread("restart", domain::restartService);
+ restarting.join(stopBoundMS());
+ assertFalse(restarting.isAlive(), "the restart of the session waits for window credit");
+ writer.join(stopBoundMS());
+ assertEquals(second.get(), ResultCode.SUCCESS);
+
+ waitForConnection(domain);
+ rs.waitForUpdates(2);
+ final List<CSN> csns = rs.receivedUpdateCSNs();
+ assertTrue(csns.get(1).isNewerThan(csns.get(0)) && csns.get(1).getServerId() == DS_ID,
+ "the change the stop gave up was not republished: " + csns);
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
+ "the replica offline message was written after a change the stop gave up");
+ waitForTheRecoveryToEnd();
+
+ rs.giveCredit(1);
+ assertEquals(replaceDescription("third"), ResultCode.SUCCESS);
+ rs.waitForUpdates(3);
+ }
+ finally
+ {
+ cleanUp(rs, restarting, writer);
+ }
+ }
+
+ /**
+ * A stop gives up only what finds no credit: with the window open, the replica offline message
+ * still goes out. And once started again, the broker waits for credit as it did before it was ever
+ * stopped: a stop does not leave behind a broker which gives a publish up as soon as the window is
+ * closed, nor one whose session is closed once the stop is past the deadline it had.
+ */
+ @Test(timeOut = 120_000)
+ public void aBrokerStartedAgainWaitsForCredit() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ Thread writer = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ domain.disable();
+ final long stopped = System.nanoTime();
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
+ domain.enable();
+ waitForConnection(domain);
+
+ assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
+ rs.waitForUpdates(1);
+
+ final AtomicReference<ResultCode> second = new AtomicReference<>();
+ writer = startThread("writer", () -> second.set(replaceDescription("second")));
+ // Two probes: the publish waited for credit twice, and did not give up after the first wait.
+ rs.waitForWindowProbes(2);
+ /* Nor past the time the stop gave the replication server to give credit and then to read: a
+ * flag the stop left would end the wait by then, and a close the stop left due the session. */
+ final long pastTheStopMS = 2 * MultimasterReplication.getConnectionTimeoutMS() + 1000
+ - NANOSECONDS.toMillis(System.nanoTime() - stopped);
+ Thread.sleep(Math.max(0, pastTheStopMS));
+ assertTrue(writer.isAlive(), "a publish gave the change up while nobody stops the broker");
+ assertEquals(rs.sessionsStarted(), 2, "the session of the broker started again was closed");
+
+ rs.giveCredit(1);
+ rs.waitForUpdates(2);
+ writer.join(stopBoundMS());
+ assertFalse(writer.isAlive(), "the local change still waits once the credit came");
+ assertEquals(second.get(), ResultCode.SUCCESS);
+ }
+ finally
+ {
+ cleanUp(rs, writer);
+ }
+ }
+
+ /**
+ * A stop ends with the broker stopped even when publishing the replica offline message fails:
+ * a broker left running as a stopping one would give up every publish which waits for credit.
+ */
+ @Test(timeOut = 120_000)
+ public void aStopWhichCannotPublishTheOfflineMessageStillStopsTheBroker() throws Exception
+ {
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ rs.start();
+ final RuntimeException failure = new IllegalStateException("cannot publish the replica offline message");
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ throw failure;
+ }
+ }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
+ try
+ {
+ connect(broker, 0);
+ try
+ {
+ broker.stop();
+ fail("the failure of the replica offline message was swallowed");
+ }
+ catch (IllegalStateException e)
+ {
+ assertSame(e, failure);
+ }
+ assertFalse(broker.isConnected(), "the broker still runs its session after it was stopped");
+ }
+ finally
+ {
+ rs.shutdown();
+ }
+ }
+
+ /**
+ * Once a publish gave its message up for the stop, the next one is refused even with the window
+ * open: the stop publishes a change which finds no credit by its deadline, then waits for the
+ * credit the replication server gives right after, and publishes a second change, which a broker
+ * that only gave up the first would write. Nor do the messages of a total update, which need no
+ * credit, go out: an import would be told it was done with an entry missing.
+ */
+ @Test(timeOut = 120_000)
+ public void aStopWritesNothingAfterAGiveUpEvenWithCredit() throws Exception
+ {
+ // The stop waits for the case to read the credit: well before the session is closed for it.
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ rs.start();
+ final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
+ final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
+ final AtomicBoolean publishedWithCredit = new AtomicBoolean();
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ final ReplicationBroker stopping = brokerRef.get();
+ stopping.publish(newDeleteMsg(csns));
+ try
+ {
+ rs.giveCredit(1);
+ readTheCredit(stopping);
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(e);
+ }
+ stopping.publish(newDeleteMsg(csns));
+ stopping.publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID, new byte[16], 1));
+ stopping.publish(new DoneMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID));
+ publishedWithCredit.set(true);
+ }
+ }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
+ brokerRef.set(broker);
+ Thread stopping = null;
+ try
+ {
+ // A read of the session which finds nothing returns in a while rather than never.
+ connect(broker, 200);
+ // The only credit of the window goes to a first change.
+ broker.publish(newDeleteMsg(csns));
+ rs.waitForUpdates(1);
+
+ stopping = startThread("stop", broker::stop);
+ stopping.join(stopBoundMS());
+ assertFalse(stopping.isAlive(), "stop() waits for window credit a replication server which takes no update does not give");
+ assertTrue(publishedWithCredit.get(), "the stop did not get to publish a change with the window open");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(LDAPUpdateMsg.class), 1, "a change was written after one the stop gave up");
+ assertEquals(rs.count(EntryMsg.class) + rs.count(DoneMsg.class), 0,
+ "a message of a total update was written after a change the stop gave up");
+ }
+ finally
+ {
+ cleanUp(rs, stopping);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * A stop which overlaps one in progress keeps what that one gave up: once the first stop gave a
+ * change up, it writes nothing more, even with the window open again, while a second stop runs
+ * beside it - which a stop of its own, with no give-up yet, would let out.
+ */
+ @Test(timeOut = 120_000)
+ public void aStopWhichOverlapsOneInProgressKeepsItsGiveUp() throws Exception
+ {
+ // The first stop waits for the case to read the credit: well before the session is closed for it.
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ rs.start();
+ final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
+ final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
+ final AtomicInteger stops = new AtomicInteger();
+ final CountDownLatch secondStopRuns = new CountDownLatch(1);
+ final CountDownLatch firstStopIsDone = new CountDownLatch(1);
+ final AtomicReference<Thread> secondStop = new AtomicReference<>();
+ final AtomicBoolean publishedWithCredit = new AtomicBoolean();
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ if (stops.incrementAndGet() > 1)
+ {
+ // The second stop waits beside the first until that one is done.
+ secondStopRuns.countDown();
+ await(firstStopIsDone);
+ return;
+ }
+ try
+ {
+ final ReplicationBroker stopping = brokerRef.get();
+ stopping.publish(newDeleteMsg(csns));
+ secondStop.set(startThread("second stop", stopping::stop));
+ await(secondStopRuns);
+ rs.giveCredit(1);
+ readTheCredit(stopping);
+ stopping.publish(newDeleteMsg(csns));
+ publishedWithCredit.set(true);
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(e);
+ }
+ finally
+ {
+ firstStopIsDone.countDown();
+ }
+ }
+ }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
+ brokerRef.set(broker);
+ Thread stopping = null;
+ try
+ {
+ // A read of the session which finds nothing returns in a while rather than never.
+ connect(broker, 200);
+ // The only credit of the window goes to a first change.
+ broker.publish(newDeleteMsg(csns));
+ rs.waitForUpdates(1);
+
+ stopping = startThread("first stop", broker::stop);
+ stopping.join(stopBoundMS());
+ assertFalse(stopping.isAlive(), "the first stop did not end");
+ assertTrue(publishedWithCredit.get(), "the first stop did not get to publish a change with the window open");
+ secondStop.get().join(stopBoundMS());
+ assertFalse(secondStop.get().isAlive(), "the second stop did not end");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(LDAPUpdateMsg.class), 1,
+ "a change was written after one the stop gave up, once a second stop had begun");
+ }
+ finally
+ {
+ cleanUp(rs, stopping, secondStop.get());
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * A replication server which keeps the session up but reads nothing of it holds the write of a
+ * local change in the socket once the buffers are full, under the lock of the pending changes
+ * stop() needs, and with credit to spare there is no wait for credit for the stop to give up: the
+ * disable has to end the write by closing the session - once the replication server had the
+ * connection timeout twice over - and the change comes back with the recovery of the next session.
+ */
+ @Test(timeOut = 120_000)
+ public void aDisableEndsAWriteToAReplicationServerWhichReadsNothing() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
+ final AtomicBoolean keepWriting = new AtomicBoolean(true);
+ final AtomicInteger changes = new AtomicInteger();
+ final AtomicReference<ResultCode> failed = new AtomicReference<>();
+ Thread writer = null;
+ Thread disabling = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs);
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+ rs.stopReading();
+ writer = startLargeChanges(keepWriting, changes, failed);
+ waitForAWriteToBlock(writer, changes);
+ keepWriting.set(false);
+
+ final long beforeTheDisable = System.nanoTime();
+ disabling = startThread("disable", domain::disable);
+ disabling.join(stopBoundMS());
+ final long disableMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable);
+ assertFalse(disabling.isAlive(), "disable() waits for a write the replication server does not read");
+ assertTrue(disableMS >= 2 * LONG_CONNECTION_TIMEOUT_MS, "disable() gave the replication server "
+ + disableMS + " ms, not twice the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
+ // Half a connection timeout above: the bound of the join leaves room for three times it.
+ assertTrue(disableMS < 2 * LONG_CONNECTION_TIMEOUT_MS + LONG_CONNECTION_TIMEOUT_MS / 2,
+ "disable() gave the replication server " + disableMS
+ + " ms, more than two and a half times the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
+ // The disable joined the listener, whose read the close of the session ended.
+ assertEquals(badDisconnectionsLogged(), 0,
+ "the close of the session by the stop was reported as one by the replication server");
+ writer.join(stopBoundMS());
+ assertFalse(writer.isAlive(), "the local change is still blocked in its write");
+ assertNull(failed.get(), "a local change failed");
+
+ // The change cut short ends what the replication server reads, and nothing comes after it.
+ rs.resumeReadingTheSessionTheDomainClosed();
+ rs.waitForTheSessionToEnd();
+ final List<CSN> beforeTheEnable = rs.receivedUpdateCSNs();
+ assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
+ "the replica offline message was written after a change the stop gave up");
+ assertEquals(rs.sessionsStarted(), 1, "a session was opened again while the domain was being stopped");
+
+ /* The disable left the state of the domain in its backend only, and the enable loads it: its
+ * newest change of this replica is the one the stop gave up, as no other change came after it. */
+ domain.enable();
+ waitForConnection(domain);
+ final CSN blocked = domain.getServerState().getCSN(DS_ID);
+ assertNotNull(blocked, "the domain has no change of its own in its state");
+ assertFalse(beforeTheEnable.contains(blocked), "the change blocked in its write was written: " + beforeTheEnable);
+ newTimer().repeatUntilSuccess(() -> assertTrue(rs.receivedUpdateCSNs().contains(blocked),
+ "the change the stop gave up, " + blocked + ", was not republished: the replication server has "
+ + rs.receivedUpdateCSNs()));
+ }
+ finally
+ {
+ cleanUp(rs, disabling, writer);
+ setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
+ }
+ }
+
+ /**
+ * The write which blocks can be one of the stop itself, with no lock of the pending changes in
+ * the way: the stop ends it the same way. The close of the session wakes the listener, which then
+ * waits for the end of the stop rather than opens a session again: the broker writes nothing more
+ * once the stop gave up, and the stop would wait for that connection.
+ */
+ @Test(timeOut = 120_000)
+ public void aStopEndsItsOwnWriteAndOpensNoSessionAgain() throws Exception
+ {
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
+ rs.start();
+ final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
+ final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
+ final AtomicReference<Thread> listenerRef = new AtomicReference<>();
+ final AtomicInteger entries = new AtomicInteger();
+ final AtomicInteger sessionsWhenTheListenerWaits = new AtomicInteger(-1);
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ final ReplicationBroker stopping = brokerRef.get();
+ // Entries need no credit: they are written until one blocks, and the stop ends that one.
+ long publishMS;
+ do
+ {
+ final long beforeThePublish = System.nanoTime();
+ stopping.publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID,
+ new byte[LARGE_MESSAGE_SIZE], entries.incrementAndGet()));
+ publishMS = NANOSECONDS.toMillis(System.nanoTime() - beforeThePublish);
+ }
+ while (publishMS < SHORT_CONNECTION_TIMEOUT_MS && entries.get() < MAX_LARGE_MESSAGES);
+ try
+ {
+ newTimer().repeatUntilSuccess(() -> assertTrue(rs.sessionsStarted() > 1
+ || waitsIn(listenerRef.get(), ReplicationBroker.class, "reStart"),
+ "the listener neither opened a session again nor waits for the end of the stop"));
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(e);
+ }
+ sessionsWhenTheListenerWaits.set(rs.sessionsStarted());
+ stopping.publish(newDeleteMsg(csns));
+ }
+ }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
+ brokerRef.set(broker);
+ Thread listener = null;
+ Thread stopping = null;
+ try
+ {
+ connect(broker, 0);
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+ listener = startThread("listener", () -> {
+ try
+ {
+ while (broker.receive() != null)
+ {
+ // What the replication server sends needs no answer here.
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ });
+ listenerRef.set(listener);
+ rs.stopReading();
+
+ final long beforeTheStop = System.nanoTime();
+ stopping = startThread("stop", broker::stop);
+ stopping.join(stopBoundMS());
+ final long stopMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheStop);
+ assertFalse(stopping.isAlive(), "stop() waits for a write the replication server does not read");
+ assertTrue(entries.get() < MAX_LARGE_MESSAGES, "no write blocked after " + entries.get() + " entries");
+ assertTrue(stopMS >= 2 * SHORT_CONNECTION_TIMEOUT_MS, "stop() gave the replication server " + stopMS
+ + " ms, not twice the connection timeout of " + SHORT_CONNECTION_TIMEOUT_MS + " ms");
+ assertEquals(sessionsWhenTheListenerWaits.get(), 1,
+ "the listener opened a session again while the broker was being stopped");
+ listener.join(stopBoundMS());
+ assertFalse(listener.isAlive(), "the listener did not end with the stop");
+ assertEquals(badDisconnectionsLogged(), 0,
+ "the close of the session by the stop was reported as one by the replication server");
+
+ rs.resumeReadingTheSessionTheDomainClosed();
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(LDAPUpdateMsg.class), 0, "a change was written after the entry the stop gave up");
+ assertFalse(broker.isConnected(), "the broker still runs its session after it was stopped");
+ }
+ finally
+ {
+ cleanUp(rs, stopping, listener);
+ }
+ }
+
+ /**
+ * A connection under way when the watchdog of the stop gives up leaves it no session to close: the
+ * listener lost the session in the stop, and connects again holding the lock of the connection
+ * phase, while replication servers which take a connection but answer none cost it the connection
+ * timeout each. A publish of the stop which waited for that lock must not write to the session the
+ * connection then opens: one which the replication server does not read would hold the stop, with
+ * nothing left to end it. The entry here goes to a replication server which reads, and must not be
+ * written at all.
+ */
+ @Test(timeOut = 120_000)
+ public void aConnectionUnderWayWhenTheStopGivesUpGetsNothingWritten() throws Exception
+ {
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
+ rs.start();
+ final List<SilentReplicationServer> silentServers = new ArrayList<>();
+ final SortedSet<String> urls = new TreeSet<>(Collections.singleton("127.0.0.1:" + rs.getPort()));
+ for (int i = 0; i < SILENT_SERVERS; i++)
+ {
+ final SilentReplicationServer silent = new SilentReplicationServer();
+ silent.start();
+ silentServers.add(silent);
+ urls.add("127.0.0.1:" + silent.getPort());
+ }
+ final DomainFakeCfg config = new DomainFakeCfg(baseDN, OTHER_DS_ID, urls);
+ config.setHeartbeatInterval(100000);
+ final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
+ final AtomicReference<Thread> listenerRef = new AtomicReference<>();
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ try
+ {
+ for (SilentReplicationServer silent : silentServers)
+ {
+ silent.answerNothing();
+ }
+ // The replication server ends the session in order, and the listener connects again.
+ rs.closeTheSession();
+ newTimer().repeatUntilSuccess(() -> assertTrue(
+ isIn(listenerRef.get(), ReplicationBroker.class, "collectReplicationServersInfo"),
+ "the listener does not connect again"));
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(e);
+ }
+ // Waits for the lock the connection holds.
+ brokerRef.get().publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID, new byte[16], 1));
+ }
+ }, new ServerState(), config, getReplSessionSecurity());
+ brokerRef.set(broker);
+ Thread listener = null;
+ Thread stopping = null;
+ try
+ {
+ connect(broker, 0);
+ listener = startThread("listener", () -> {
+ try
+ {
+ while (broker.receive() != null)
+ {
+ // What the replication server sends needs no answer here.
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ });
+ listenerRef.set(listener);
+
+ final long beforeTheStop = System.nanoTime();
+ final Thread stop = startThread("stop", broker::stop);
+ stopping = stop;
+ newTimer().repeatUntilSuccess(() -> assertTrue(
+ stop.getState() == Thread.State.BLOCKED && isIn(stop, ReplicationBroker.class, "publish"),
+ "the entry of the stop does not wait for the lock of the connection phase"));
+ // Past the deadline of the watchdog, which then found the connection still under way.
+ Thread.sleep(Math.max(0, 2 * SHORT_CONNECTION_TIMEOUT_MS + 500
+ - NANOSECONDS.toMillis(System.nanoTime() - beforeTheStop)));
+ assertFalse(watchdogRuns(), "the watchdog of the stop did not act at its deadline");
+ assertTrue(isIn(listener, ReplicationBroker.class, "connectAsDataServer"),
+ "the connection ended before the watchdog of the stop acted: the case tests nothing");
+
+ stop.join(stopBoundMS());
+ assertFalse(stop.isAlive(), "stop() did not end once the connection did");
+ assertEquals(rs.sessionsStarted(), 2, "the connection under way did not open its session");
+ rs.waitForTheSessionToEnd();
+ assertEquals(rs.count(EntryMsg.class), 0,
+ "the stop wrote to the session a connection opened once the stop had given up");
+ listener.join(stopBoundMS());
+ assertFalse(listener.isAlive(), "the listener did not end with the stop");
+ }
+ finally
+ {
+ for (SilentReplicationServer silent : silentServers)
+ {
+ silent.shutdown();
+ }
+ cleanUp(rs, stopping, listener);
+ }
+ }
+
+ /**
+ * A give-up for credit closes nothing: a replication server which then drops the session is
+ * reported as one which did, as it is with no stop at all. Only a close this server makes - by the
+ * watchdog of a stop, or by the heartbeat monitor - is not reported, as the close is marked begun
+ * before the sockets are closed.
+ */
+ @Test(timeOut = 120_000)
+ public void aDropByTheReplicationServerAfterAGiveUpIsStillReported() throws Exception
+ {
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer();
+ rs.start();
+ final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
+ final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
+ final AtomicReference<Thread> listenerRef = new AtomicReference<>();
+ final AtomicInteger reportedInTheStop = new AtomicInteger(-1);
+ final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
+ {
+ @Override
+ public void publishReplicaOfflineMsg()
+ {
+ // No credit: the change is given up at the credit deadline of the stop.
+ brokerRef.get().publish(newDeleteMsg(csns));
+ try
+ {
+ rs.dropTheSession();
+ // The listener reports what ended its read before it waits for the end of the stop.
+ newTimer().repeatUntilSuccess(() -> assertTrue(waitsIn(listenerRef.get(), ReplicationBroker.class, "reStart"),
+ "the listener does not wait for the end of the stop"));
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(e);
+ }
+ reportedInTheStop.set(badDisconnectionsLogged());
+ }
+ }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
+ brokerRef.set(broker);
+ Thread listener = null;
+ Thread stopping = null;
+ try
+ {
+ connect(broker, 0);
+ listener = startThread("listener", () -> {
+ try
+ {
+ while (broker.receive() != null)
+ {
+ // What the replication server sends needs no answer here.
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ });
+ listenerRef.set(listener);
+ // The only credit of the window goes to a first change.
+ broker.publish(newDeleteMsg(csns));
+ rs.waitForUpdates(1);
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+
+ stopping = startThread("stop", broker::stop);
+ stopping.join(stopBoundMS());
+ assertFalse(stopping.isAlive(), "stop() did not end");
+ assertTrue(reportedInTheStop.get() > 0,
+ "the replication server dropped the session after a give-up for credit, and nothing reported it");
+ assertEquals(rs.count(LDAPUpdateMsg.class), 1, "the change was written with no credit");
+ }
+ finally
+ {
+ cleanUp(rs, stopping, listener);
+ }
+ }
+
+ /**
+ * A replication server which is gone without a reset reads nothing, sends no heartbeat and answers
+ * no connection. Once a write to it blocks, the heartbeat monitor still has to close the session,
+ * not wait for that write, which nothing but TCP would end: the change then ends, and once the
+ * replication server is back, the domain connects again and its changes reach it, the one cut
+ * short included.
+ */
+ @Test(timeOut = 120_000)
+ public void theHeartbeatMonitorEndsAWriteToAReplicationServerWhichIsGone() throws Exception
+ {
+ TestCaseUtils.initializeTestBackend(true);
+ final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
+ final AtomicBoolean keepWriting = new AtomicBoolean(true);
+ final AtomicInteger changes = new AtomicInteger();
+ final AtomicReference<ResultCode> failed = new AtomicReference<>();
+ Thread writer = null;
+ try
+ {
+ final LDAPReplicationDomain domain = startDomain(rs, HEARTBEAT_INTERVAL_MS);
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+ rs.goAway();
+ final long away = System.nanoTime();
+ writer = startLargeChanges(keepWriting, changes, failed);
+ waitForAWriteToBlock(writer, changes);
+ keepWriting.set(false);
+ final long blocked = System.nanoTime();
+
+ // The monitor gives up two or three intervals after the last heartbeat.
+ writer.join(4 * HEARTBEAT_INTERVAL_MS + 10000);
+ final long released = System.nanoTime();
+ assertFalse(writer.isAlive(), "the write still blocks once the heartbeats of the replication server stopped");
+ assertTrue(NANOSECONDS.toMillis(released - away) >= 2 * HEARTBEAT_INTERVAL_MS, "the write of change "
+ + changes.get() + ", blocked " + NANOSECONDS.toMillis(blocked - away) + " ms after the replication server went away, ended "
+ + NANOSECONDS.toMillis(released - away) + " ms after it: before the heartbeat monitor could give the session up");
+ assertNull(failed.get(), "a local change failed");
+
+ rs.resumeReading();
+ waitForConnection(domain);
+ final CSN last = domain.getServerState().getCSN(DS_ID);
+ assertNotNull(last, "the domain has no change of its own in its state");
+ newTimer().repeatUntilSuccess(() -> assertTrue(rs.receivedUpdateCSNs().contains(last),
+ "the change cut short, " + last + ", did not reach the replication server: it has "
+ + rs.receivedUpdateCSNs()));
+ assertTrue(rs.sessionsStarted() > 1, "the domain kept the session whose heartbeats stopped");
+ // The listener connected again: it is past what it reports of the read the close ended.
+ assertEquals(badDisconnectionsLogged(), 0,
+ "the close of the session by the heartbeat monitor was reported as one by the replication server");
+ }
+ finally
+ {
+ cleanUp(rs, writer);
+ }
+ }
+
+ private LDAPReplicationDomain startDomain(WithholdingReplicationServer rs) throws Exception
+ {
+ // Heartbeats far apart: the session stays up whatever the replication server reads.
+ return startDomain(rs, 100000);
+ }
+
+ private LDAPReplicationDomain startDomain(WithholdingReplicationServer rs, long heartbeatIntervalMS)
+ throws Exception
+ {
+ rs.start();
+ final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID,
+ new TreeSet<>(Collections.singleton("127.0.0.1:" + rs.getPort())));
+ conf.setHeartbeatInterval(heartbeatIntervalMS);
+ final LDAPReplicationDomain domain = MultimasterReplication.createNewDomain(conf);
+ domain.start();
+ waitForConnection(domain);
+ return domain;
+ }
+
+ private void waitForConnection(final LDAPReplicationDomain domain) throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "the domain is not connected"));
+ }
+
+ /**
+ * Waits for the thread which publishes the changes the replication server misses to end: until
+ * then the domain does not push its own changes, but leaves them to that thread.
+ */
+ private static void waitForTheRecoveryToEnd() throws Exception
+ {
+ final String name = "Replica DS(" + DS_ID + ") missing change publisher";
+ newTimer().repeatUntilSuccess(() -> {
+ for (Thread thread : Thread.getAllStackTraces().keySet())
+ {
+ assertFalse(thread.isAlive() && thread.getName().startsWith(name), "the recovery still runs");
+ }
+ });
+ }
+
+ /** Sets the connection timeout of the replication provider, which the stop of a broker reads. */
+ private static void setConnectionTimeout(long timeoutMS)
+ {
+ assertEquals(getRootConnection().processModify(
+ modifyRequest(PROVIDER_DN, REPLACE, "ds-cfg-connection-timeout", timeoutMS + " ms")).getResultCode(),
+ ResultCode.SUCCESS);
+ assertEquals(MultimasterReplication.getConnectionTimeoutMS(), timeoutMS);
+ }
+
+ private DeleteMsg newDeleteMsg(CSNGenerator csns)
+ {
+ final CSN csn = csns.newCSN();
+ return new DeleteMsg(DN.valueOf("cn=" + csn + "," + baseDN), csn, "entryUUID-" + csn);
+ }
+
+ /** No domain reads the session of the broker provided: the credit counts once the case reads it. */
+ private static void readTheCredit(final ReplicationBroker broker) throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> {
+ try
+ {
+ broker.receive();
+ }
+ catch (SocketTimeoutException e)
+ {
+ // Nothing more to read.
+ }
+ assertEquals(broker.getCurrentSendWindow(), 1, "the credit the replication server gave");
+ });
+ }
+
+ private static void await(final CountDownLatch latch)
+ {
+ try
+ {
+ assertTrue(latch.await(30, SECONDS), "a thread of the case did not get where the case waits for it");
+ }
+ catch (InterruptedException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /** Waits for stop() to wait for the lock of the pending changes, which a local change holds. */
+ private static void waitForTheOfflineMessageToWaitForTheLock(final Thread disabling) throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertTrue(disabling.getState() == Thread.State.BLOCKED
+ && isIn(disabling, PendingChanges.class, "putReplicaOfflineMsg"),
+ "stop() does not wait for the lock of the pending changes"));
+ }
+
+ /**
+ * Returns how many times the error log says, since it was cleared, that the session was closed by
+ * the replication server rather than by this server: what the listener reports when a read fails
+ * on a session whose close it does not find begun.
+ */
+ private static int badDisconnectionsLogged()
+ {
+ int count = 0;
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ count += record.contains("msgID=" + WARN_REPLICATION_SERVER_BADLY_DISCONNECTED.ordinal() + " ") ? 1 : 0;
+ }
+ return count;
+ }
+
+ /**
+ * Starts a thread which makes changes of {@link #LARGE_MESSAGE_SIZE} until it is told to stop,
+ * counting them, and keeping the first result which is no success.
+ */
+ private Thread startLargeChanges(final AtomicBoolean keepWriting, final AtomicInteger changes,
+ final AtomicReference<ResultCode> failed)
+ {
+ return startThread("writer", () -> {
+ while (keepWriting.get() && changes.get() < MAX_LARGE_MESSAGES)
+ {
+ final ResultCode result = replaceDescription(changes.get() + LARGE_VALUE);
+ if (result != ResultCode.SUCCESS)
+ {
+ failed.compareAndSet(null, result);
+ }
+ changes.incrementAndGet();
+ }
+ });
+ }
+
+ /**
+ * Waits for the writer to stay in a write to the session, the number of the changes it made
+ * standing still: the socket buffers are full, and the replication server reads nothing.
+ */
+ private static void waitForAWriteToBlock(final Thread writer, final AtomicInteger changes) throws Exception
+ {
+ // Two seconds in a row, so that a pause of the collector in the middle of a write is not taken for it.
+ new TestTimer.Builder().maxSleep(20, SECONDS).sleepTimes(50, MILLISECONDS).toTimer().repeatUntilSuccess(() -> {
+ assertTrue(writer.isAlive(), "no write blocked after " + changes.get() + " changes");
+ final int before = changes.get();
+ for (int second = 0; second < 2; second++)
+ {
+ Thread.sleep(1000);
+ assertTrue(changes.get() == before && isIn(writer, Session.class, "send"),
+ "no write blocks after " + changes.get() + " changes");
+ }
+ });
+ }
+
+ /** Tells whether the watchdog of a stop of the broker of {@link #OTHER_DS_ID} runs. */
+ private static boolean watchdogRuns()
+ {
+ final String name = "Replica DS(" + OTHER_DS_ID + ") watchdog of the stop";
+ for (Thread thread : Thread.getAllStackTraces().keySet())
+ {
+ if (thread.isAlive() && thread.getName().startsWith(name))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Tells whether the thread waits, for a lock or for a time, inside the method provided. */
+ private static boolean waitsIn(final Thread thread, final Class<?> type, final String method)
+ {
+ final Thread.State state = thread.getState();
+ return (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) && isIn(thread, type, method);
+ }
+
+ private static boolean isIn(final Thread thread, final Class<?> type, final String method)
+ {
+ for (StackTraceElement frame : thread.getStackTrace())
+ {
+ if (frame.getClassName().equals(type.getName()) && frame.getMethodName().equals(method))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns the CSN of the announcement that the replica of the domain went offline which the
+ * shutdown of a collocated replication server would wait for the forward of, if there is one.
+ */
+ private CSN offlineAnnouncementOfTheReplica() throws Exception
+ {
+ final Field shutdownSync = MultimasterReplication.class.getDeclaredField("dsrsShutdownSync");
+ shutdownSync.setAccessible(true);
+ final Field announcements = DSRSShutdownSync.class.getDeclaredField("replicaOfflineMsgs");
+ announcements.setAccessible(true);
+ final Map<?, ?> byDomain = (Map<?, ?>) announcements.get(shutdownSync.get(null));
+ final Map<?, ?> byReplica = (Map<?, ?>) byDomain.get(baseDN);
+ final Object announcement = byReplica != null ? byReplica.get(DS_ID) : null;
+ if (announcement == null)
+ {
+ return null;
+ }
+ final Field csn = announcement.getClass().getDeclaredField("csn");
+ csn.setAccessible(true);
+ return (CSN) csn.get(announcement);
+ }
+
+ private ResultCode replaceDescription(String value)
+ {
+ return getRootConnection().processModify(modifyRequest(baseDN, REPLACE, "description", value))
+ .getResultCode();
+ }
+
+ /**
+ * How long disabling the domain may take: the stop gives the replication server as long as a
+ * connection to it to give credit, and as long again to read what was written, and the bound adds
+ * a margin far beyond what the stop needs on top of that, and far below what a stop which waits
+ * for the replication server would take: forever. It is also far below the 30 seconds of
+ * connection timeout the suite runs with.
+ */
+ private static long stopBoundMS()
+ {
+ return 2 * MultimasterReplication.getConnectionTimeoutMS() + 10000;
+ }
+
+ private static Thread startThread(String name, Runnable runnable)
+ {
+ final Thread thread = new Thread(runnable, ReplicationBrokerStopTest.class.getSimpleName() + " " + name);
+ thread.setDaemon(true);
+ thread.start();
+ return thread;
+ }
+
+ /**
+ * Closes the replication server and then waits for the threads: on a broker which waits for credit
+ * whatever happens, only the end of the session ends their wait - the broker finds no replication
+ * server to connect to again and gives the publish up.
+ */
+ private void cleanUp(WithholdingReplicationServer rs, Thread... threads) throws Exception
+ {
+ rs.shutdown();
+ for (Thread thread : threads)
+ {
+ if (thread != null)
+ {
+ thread.join(30000);
+ }
+ }
+ // A no-op when no domain was registered.
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+
+ private static TestTimer newTimer()
+ {
+ return new TestTimer.Builder().maxSleep(10, SECONDS).sleepTimes(50, MILLISECONDS).toTimer();
+ }
+
+ /**
+ * A replication server which keeps the session of the domain up and reads every message of it,
+ * but gives no credit beyond the window it opens the session with - of one update, unless the
+ * test opens a larger one - unless the test asks it to. Like a replication server, it reports the
+ * changes it received as its state: a domain which finds itself ahead of it republishes what it
+ * misses. It sends the heartbeats the domain asks for, stops reading its sessions - or goes away
+ * altogether, heartbeats included - when the test asks it to, and serves each connection on a
+ * thread of its own, as a replication server does: a session it does not read does not keep the
+ * next connection waiting.
+ */
+ private static final class WithholdingReplicationServer extends Thread
+ {
+ private static final int RS_ID = 2;
+
+ private final int window;
+ private final ServerSocket listenSocket;
+ /** The latest session started, until it ends. */
+ private final AtomicReference<Session> session = new AtomicReference<>();
+ /** Every session being served, the handshakes included, for the shutdown to close. */
+ private final List<Session> sessions = new CopyOnWriteArrayList<>();
+ private final List<Thread> servers = new CopyOnWriteArrayList<>();
+ private final AtomicInteger sessionsStarted = new AtomicInteger();
+ /** The kinds of the messages received on the sessions, in the order they came. */
+ private final List<Class<?>> received = new CopyOnWriteArrayList<>();
+ /** The CSNs of the changes received on the sessions, in the order they came. */
+ private final List<CSN> updateCSNs = new CopyOnWriteArrayList<>();
+ private final ServerState state = new ServerState();
+ /** Guards {@link #reading} and the changes of {@link #away}. */
+ private final Object readingLock = new Object();
+ /** Whether the sessions read what the domain writes: each reads one message more once it is not. */
+ private boolean reading = true;
+ /** Whether this server is away: it then sends no heartbeat, and answers no connection either. */
+ private volatile boolean away;
+
+ WithholdingReplicationServer() throws IOException
+ {
+ this(1);
+ }
+
+ WithholdingReplicationServer(int window) throws IOException
+ {
+ super(ReplicationBrokerStopTest.class.getSimpleName() + " replication server");
+ setDaemon(true);
+ this.window = window;
+ listenSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+ }
+
+ int getPort()
+ {
+ return listenSocket.getLocalPort();
+ }
+
+ @Override
+ public void run()
+ {
+ while (!listenSocket.isClosed())
+ {
+ final Socket socket;
+ try
+ {
+ socket = listenSocket.accept();
+ }
+ catch (IOException e)
+ {
+ // The test closed this server.
+ continue;
+ }
+ final Thread server = new Thread(() -> serve(socket), getName() + " session");
+ server.setDaemon(true);
+ servers.add(server);
+ server.start();
+ }
+ }
+
+ /**
+ * Answers both phases of the handshake: the first connection of the domain only asks for the
+ * state of this server and ends with a StopMsg, the one it keeps goes on to start a session.
+ */
+ private void serve(Socket socket)
+ {
+ Session s = null;
+ try
+ {
+ // A server which is away answers no connection: the connection waits for it to come back.
+ waitToBeBack();
+ s = new ReplSessionSecurity().createServerSession(socket, MultimasterReplication.getConnectionTimeoutMS());
+ sessions.add(s);
+ final ServerStartMsg start = (ServerStartMsg) s.receive();
+ final String url = "127.0.0.1:" + getPort();
+ s.publish(new ReplServerStartMsg(RS_ID, url, start.getBaseDN(), window, state.duplicate(),
+ start.getGenerationId(), start.getSSLEncryption(), start.getGroupId(), 5000));
+ if (!start.getSSLEncryption())
+ {
+ s.stopEncryption();
+ }
+ if (!(s.receive() instanceof StartSessionMsg))
+ {
+ return;
+ }
+ sessionsStarted.incrementAndGet();
+ s.publish(new TopologyMsg(new ArrayList<DSInfo>(), Collections.singletonList(
+ new RSInfo(RS_ID, url, start.getGenerationId(), start.getGroupId(), 1))));
+ s.setSoTimeout(0);
+ session.set(s);
+ startHeartbeats(s, start.getHeartbeatInterval());
+ while (true)
+ {
+ waitToRead();
+ final ReplicationMsg msg = s.receive();
+ if (msg instanceof LDAPUpdateMsg)
+ {
+ final CSN csn = ((LDAPUpdateMsg) msg).getCSN();
+ updateCSNs.add(csn);
+ state.update(csn);
+ }
+ received.add(msg.getClass());
+ }
+ }
+ catch (Exception e)
+ {
+ // The domain closed the session, or the test closed this server.
+ }
+ finally
+ {
+ if (s != null)
+ {
+ session.compareAndSet(s, null);
+ s.close();
+ sessions.remove(s);
+ }
+ }
+ }
+
+ private void waitToRead() throws InterruptedException
+ {
+ synchronized (readingLock)
+ {
+ while (!reading)
+ {
+ readingLock.wait();
+ }
+ }
+ }
+
+ private void waitToBeBack() throws InterruptedException
+ {
+ synchronized (readingLock)
+ {
+ while (away)
+ {
+ readingLock.wait();
+ }
+ }
+ }
+
+ /**
+ * Sends heartbeats on the session at the interval the domain asked for, as a replication server
+ * does from a thread of its own, until the session ends - and none while this server is away.
+ */
+ private void startHeartbeats(final Session s, final long intervalMS)
+ {
+ if (intervalMS <= 0)
+ {
+ return;
+ }
+ final Thread heartbeats = new Thread(() -> {
+ try
+ {
+ while (!s.closeInitiated())
+ {
+ Thread.sleep(intervalMS / 2);
+ if (!away)
+ {
+ s.publish(new HeartbeatMsg());
+ }
+ }
+ }
+ catch (IOException | InterruptedException e)
+ {
+ // The session ended.
+ }
+ }, getName() + " heartbeats");
+ heartbeats.setDaemon(true);
+ heartbeats.start();
+ }
+
+ /** Stops reading the sessions, after the message each of them is reading; the heartbeats go on. */
+ void stopReading()
+ {
+ synchronized (readingLock)
+ {
+ reading = false;
+ }
+ }
+
+ /**
+ * Goes away as a server whose host is down does, with no reset of its connections: it reads
+ * nothing more, sends no heartbeat, and answers no new connection.
+ */
+ void goAway()
+ {
+ synchronized (readingLock)
+ {
+ away = true;
+ reading = false;
+ }
+ }
+
+ /** Reads the sessions again - and, if it was away, sends heartbeats and answers connections again. */
+ void resumeReading()
+ {
+ synchronized (readingLock)
+ {
+ away = false;
+ reading = true;
+ readingLock.notifyAll();
+ }
+ }
+
+ /**
+ * Reads the sessions again once the domain closed its session while this server read nothing,
+ * and ends that session at once. What the domain left queued at its end goes out once its TCP
+ * finds the window open again: from the update the reads here send, or else from its next window
+ * probe - and after a stall of seconds the probes are seconds apart, 13 s after a stall of 14 s
+ * on Linux, longer than the wait for the end of the session. A heartbeat the other way has the
+ * closed end answer with a reset at once: this server reads what reached it, and the read ends.
+ */
+ void resumeReadingTheSessionTheDomainClosed()
+ {
+ final Session closed = session.get();
+ resumeReading();
+ if (closed != null)
+ {
+ try
+ {
+ closed.publish(new HeartbeatMsg());
+ }
+ catch (IOException e)
+ {
+ // The reset came already.
+ }
+ }
+ }
+
+ /** Ends the latest session in order, with a StopMsg, as a replication server which stops does. */
+ void closeTheSession()
+ {
+ session.get().close();
+ }
+
+ /** Drops the latest session with no StopMsg, as a replication server which fails does. */
+ void dropTheSession()
+ {
+ session.get().abort(new IOException("the replication server drops the session"));
+ }
+
+ /** Returns how many sessions were started, the connections which only ask for the state left out. */
+ int sessionsStarted()
+ {
+ return sessionsStarted.get();
+ }
+
+ int count(Class<? extends ReplicationMsg> kind)
+ {
+ int count = 0;
+ for (Class<?> msg : received)
+ {
+ count += kind.isAssignableFrom(msg) ? 1 : 0;
+ }
+ return count;
+ }
+
+ /** Returns the kinds of the updates received, replica offline messages included, in order. */
+ List<Class<?>> receivedUpdates()
+ {
+ final List<Class<?>> updates = new ArrayList<>();
+ for (Class<?> msg : received)
+ {
+ if (LDAPUpdateMsg.class.isAssignableFrom(msg) || ReplicaOfflineMsg.class.isAssignableFrom(msg))
+ {
+ updates.add(msg);
+ }
+ }
+ return updates;
+ }
+
+ List<CSN> receivedUpdateCSNs()
+ {
+ return new ArrayList<>(updateCSNs);
+ }
+
+ void giveCredit(int credit) throws IOException
+ {
+ session.get().publish(new WindowMsg(credit));
+ }
+
+ /** Gives credit unless the domain ended the session, before this reads it or while this writes. */
+ void giveCreditIfTheSessionIsUp(int credit)
+ {
+ final Session s = session.get();
+ if (s != null)
+ {
+ try
+ {
+ s.publish(new WindowMsg(credit));
+ }
+ catch (IOException e)
+ {
+ // The domain closed the session in between.
+ }
+ }
+ }
+
+ void waitForUpdates(final int count) throws Exception
+ {
+ newTimer().repeatUntilSuccess(
+ () -> assertEquals(count(LDAPUpdateMsg.class), count, "updates the replication server received"));
+ }
+
+ void waitForWindowProbes(final int count) throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertTrue(count(WindowProbeMsg.class) >= count,
+ "window probes the replication server received"));
+ }
+
+ /**
+ * Waits for the session to end: what the domain wrote to it has all been read by then - or, once
+ * the session was ended by a reset, what of it reached this server.
+ */
+ void waitForTheSessionToEnd() throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertNull(session.get(), "the session of the domain is still up"));
+ }
+
+ void shutdown() throws Exception
+ {
+ listenSocket.close();
+ resumeReading();
+ for (Session s : sessions)
+ {
+ s.close();
+ }
+ join(30000);
+ for (Thread server : servers)
+ {
+ server.join(30000);
+ }
+ }
+ }
+
+ /**
+ * A replication server which takes each connection and, once told to, answers nothing on it, as
+ * one which hangs does: a connection to it waits for the connection timeout in its SSL
+ * negotiation. Until then it closes each connection at once, which fails it at once.
+ */
+ private static final class SilentReplicationServer extends Thread
+ {
+ private final ServerSocket listenSocket;
+ private final List<Socket> held = new CopyOnWriteArrayList<>();
+ private volatile boolean silent;
+
+ SilentReplicationServer() throws IOException
+ {
+ super(ReplicationBrokerStopTest.class.getSimpleName() + " silent replication server");
+ setDaemon(true);
+ listenSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+ }
+
+ int getPort()
+ {
+ return listenSocket.getLocalPort();
+ }
+
+ @Override
+ public void run()
+ {
+ while (!listenSocket.isClosed())
+ {
+ try
+ {
+ final Socket socket = listenSocket.accept();
+ if (silent)
+ {
+ held.add(socket);
+ }
+ else
+ {
+ socket.close();
+ }
+ }
+ catch (IOException e)
+ {
+ // The test closed this server.
+ }
+ }
+ }
+
+ void answerNothing()
+ {
+ silent = true;
+ }
+
+ void shutdown() throws Exception
+ {
+ listenSocket.close();
+ for (Socket socket : held)
+ {
+ socket.close();
+ }
+ join(30000);
+ }
+ }
+}
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