[#1202] [#1209] [#1222] Give the replication server a bounded time to give ...

Valery Kharseko

22 hours ago 77ad03e14cc1bbc5471f332ba73430d4c472b366

 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>

 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");
    }
  }


 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);
  }



  /**

 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

 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();
      }
    }
  }
}

 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"));
      }
    }
  }


 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)
  {

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java

New file
@@ -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);
    }
  }
}

 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;