[#917] Wait for every peer replication server to forward the ReplicaOffline...

Valery Kharseko

12 hours ago 2c7e382da347e11820463fcf977d885c2f40d764

[#917] Wait for every peer replication server to forward the ReplicaOfflineMsg (#947)

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java

@@ -1226,6 +1226,15 @@
   * Waits for the ReplicaOfflineMsg of every domain which has a replication server to forward it
   * to. With no such server connected there is nobody to forward the message to, and waiting
   * would only delay the shutdown by the whole grace period.
   * <p>
   * The recipients DSRSShutdownSync records when the message is queued are the sharper source of
   * truth and cover the domains this test lets through - a peer which connected after the
   * message was queued owes nothing, and is not waited for. What this test still covers is the
   * announcement which was never queued here at all, and for which no recipient can therefore be
   * recorded: the message of a replica which picked a remote replication server, or the
   * announcement of issue #918 recorded after its message was relayed. Those wait out the whole
   * grace period on the first forward, and on a server with no peer connected nothing would ever
   * report one.
   */
  private void awaitReplicaOfflineMsgsForwarded()
  {

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -361,15 +361,23 @@
    // Push the message to the replication servers
    if (sourceHandler.isDataServer())
    {
      for (ReplicationServerHandler rsHandler : connectedRSs.values())
      if (updateMsg instanceof ReplicaOfflineMsg)
      {
        /**
         * Ignore updates to RS with bad gen id
         * (no system managed status for a RS)
         */
        if (!isDifferentGenerationId(rsHandler, updateMsg))
        pushReplicaOfflineMsgToReplicationServers(
            updateMsg, notAssuredUpdateMsg, assuredServers);
      }
      else
      {
        for (ReplicationServerHandler rsHandler : connectedRSs.values())
        {
          addUpdate(rsHandler, updateMsg, notAssuredUpdateMsg, assuredServers);
          /**
           * Ignore updates to RS with bad gen id
           * (no system managed status for a RS)
           */
          if (!isDifferentGenerationId(rsHandler, updateMsg))
          {
            addUpdate(rsHandler, updateMsg, notAssuredUpdateMsg, assuredServers);
          }
        }
      }
    }
@@ -386,6 +394,54 @@
    }
  }

  /**
   * Pushes a ReplicaOfflineMsg to the replication servers which have to relay it, recording
   * which of them it goes to before any of them can forward it.
   * <p>
   * Each of them is served by its own writer, and the shutdown of a collocated directory server
   * waits for every one of them to have forwarded the message before it stops the handlers - see
   * DSRSShutdownSync. The recipients are recorded first because a forward reported before they
   * are known ends that wait at once.
   */
  private void pushReplicaOfflineMsgToReplicationServers(UpdateMsg offlineMsg,
      NotAssuredUpdateMsg notAssuredUpdateMsg, List<Integer> assuredServers)
  {
    final List<ReplicationServerHandler> recipients = new ArrayList<>(connectedRSs.size());
    final List<Integer> recipientIds = new ArrayList<>(connectedRSs.size());
    for (ReplicationServerHandler rsHandler : connectedRSs.values())
    {
      // Ignore updates to RS with bad gen id (no system managed status for a RS)
      if (!isDifferentGenerationId(rsHandler, offlineMsg))
      {
        recipients.add(rsHandler);
        recipientIds.add(rsHandler.getServerId());
      }
    }
    localReplicationServer.getDSRSShutdownSync()
        .replicaOfflineMsgDispatched(baseDN, offlineMsg.getCSN(), recipientIds);
    for (ReplicationServerHandler rsHandler : recipients)
    {
      /*
       * A recipient whose teardown started while the list was being built has had its message
       * queue cleared already, and its own give-up ran before it was recorded here: queueing for
       * it would only leave the shutdown waiting for a forward which can no longer happen.
       * <p>
       * This is read after the recipients were recorded, and every teardown - stopServer() and
       * unregisterFailedHandshake() - raises the flag before it gives up. So of the two, at
       * least one always sees the other: a teardown whose give-up came too early to find this
       * recipient has necessarily raised the flag this reads.
       */
      if (rsHandler.shuttingDown())
      {
        noLongerAwaitTheForwardOf(rsHandler);
      }
      else
      {
        addUpdate(rsHandler, offlineMsg, notAssuredUpdateMsg, assuredServers);
      }
    }
  }

  private boolean isDifferentGenerationId(ReplicationServerHandler rsHandler,
      UpdateMsg updateMsg)
  {
@@ -1068,60 +1124,89 @@
    if (!sHandler.engageShutdown())
      // Only do this once (prevent other thread to enter here again)
    {
      if (!shutdown)
      {
        try
        {
          // Acquire lock on domain (see more details in comment of start()
          // method of ServerHandler)
          lock();
        }
        catch (InterruptedException ex)
        {
          // We can't deal with this here, so re-interrupt thread so that it is
          // caught during subsequent IO.
          Thread.currentThread().interrupt();
          return;
        }
      }

      try
      {
        // Stop useless monitoring publisher if no more RS or DS in domain
        if ( (connectedDSs.size() + connectedRSs.size() )== 1)
        {
          if (logger.isTraceEnabled())
          {
            debug("remote server " + sHandler
                + " is the last RS/DS to be stopped:"
                + " stopping monitoring publisher");
          }
          stopMonitoringPublisher();
        }

        if (connectedRSs.containsKey(sHandler.getServerId()))
        {
          unregisterServerHandler(sHandler, shutdown, false);
        }
        else if (connectedDSs.containsKey(sHandler.getServerId()))
        {
          unregisterServerHandler(sHandler, shutdown, true);
        }
      }
      catch(Exception e)
      {
        logger.error(LocalizableMessage.raw(stackTraceToSingleLineString(e)));
      }
      finally
      {
        if (!shutdown)
        {
          release();
          try
          {
            // Acquire lock on domain (see more details in comment of start()
            // method of ServerHandler)
            lock();
          }
          catch (InterruptedException ex)
          {
            // We can't deal with this here, so re-interrupt thread so that it is
            // caught during subsequent IO.
            Thread.currentThread().interrupt();
            return;
          }
        }

        try
        {
          // Stop useless monitoring publisher if no more RS or DS in domain
          if ( (connectedDSs.size() + connectedRSs.size() )== 1)
          {
            if (logger.isTraceEnabled())
            {
              debug("remote server " + sHandler
                  + " is the last RS/DS to be stopped:"
                  + " stopping monitoring publisher");
            }
            stopMonitoringPublisher();
          }

          if (connectedRSs.containsKey(sHandler.getServerId()))
          {
            unregisterServerHandler(sHandler, shutdown, false);
          }
          else if (connectedDSs.containsKey(sHandler.getServerId()))
          {
            unregisterServerHandler(sHandler, shutdown, true);
          }
        }
        catch(Exception e)
        {
          logger.error(LocalizableMessage.raw(stackTraceToSingleLineString(e)));
        }
        finally
        {
          if (!shutdown)
          {
            release();
          }
        }
      }
      finally
      {
        /*
         * On every exit of this block, including the interrupted lock acquisition above and an
         * unregistration which threw: the flag engageShutdown() has just set is one-shot, so no
         * later stopServer() will run for this handler and nothing else would strike it off.
         * Every caller closed the session before getting here or is stopping the handler on
         * purpose, so this peer can no longer be told anything whatever this method managed to
         * do.
         */
        if (sHandler.isReplicationServer())
        {
          noLongerAwaitTheForwardOf(sHandler);
        }
      }
    }
  }

  /**
   * A peer replication server which is gone can no longer forward the ReplicaOfflineMsg it was
   * given, so the shutdown of a collocated directory server must stop waiting for it - see
   * DSRSShutdownSync.
   */
  void noLongerAwaitTheForwardOf(ServerHandler rsHandler)
  {
    localReplicationServer.getDSRSShutdownSync()
        .replicaOfflineMsgNotForwarded(baseDN, rsHandler.getServerId());
  }

  private void unregisterServerHandler(ServerHandler sHandler, boolean shutdown,
      boolean isDirectoryServer)
  {
@@ -1194,6 +1279,22 @@
    {
      return;
    }
    /*
     * This handler is stopping, and nothing else will ever say so for it: the reader and writer
     * threads whose death normally reaches stopServer() were never started, so no teardown of
     * its own will run. The flag has to be set before the give-up below, because that is the
     * order put() reads the two in - it records the recipients of a ReplicaOfflineMsg and only
     * then asks each of them whether it is shutting down. Skip it and a message being pushed
     * concurrently slips between the two: the give-up runs while nothing is recorded yet and
     * does nothing, the recipients are then recorded with this handler among them, and the loop
     * queues the message for a handler whose queue is about to be cleared - leaving the shutdown
     * to wait out its whole grace period for a forward nobody can report.
     */
    sHandler.engageShutdown();
    if (!isDataServer)
    {
      noLongerAwaitTheForwardOf(sHandler);
    }

    if (connectedDSs.isEmpty() && connectedRSs.isEmpty())
    {

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java

@@ -105,7 +105,22 @@
           "Connection closure: null update returned by domain.");
          break;
        }
        if (!isUpdateMsgFiltered(updateMsg))
        if (isUpdateMsgFiltered(updateMsg))
        {
          /*
           * The message is dropped here and will not be published to this server. When it is the
           * ReplicaOfflineMsg a shutdown is waiting for - its filter is wider than the one
           * ReplicationServerDomain.put() applied when it queued the message, so a peer RS can
           * be given a message this drops - the shutdown must stop waiting for a forward which
           * will never be reported.
           */
          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
          {
            dsrsShutdownSync.replicaOfflineMsgNotForwarded(
                replicationServerDomain.getBaseDN(), handler.getServerId());
          }
        }
        else
        {
          // Publish the update to the remote server using a protocol version it supports
          session.publish(updateMsg);
@@ -121,7 +136,7 @@
          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
          {
            dsrsShutdownSync.replicaOfflineMsgForwarded(
                replicationServerDomain.getBaseDN(), updateMsg.getCSN());
                replicationServerDomain.getBaseDN(), updateMsg.getCSN(), handler.getServerId());
          }
        }
      }

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java

@@ -20,6 +20,8 @@
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.Collection;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@@ -51,9 +53,13 @@
  private final long gracePeriod;

  /**
   * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per
   * The ReplicaOfflineMsg which is still owed a forward, per domain and per
   * replica of that domain.
   * <p>
   * An entry lives until every replication server the message was queued for
   * has forwarded it, so it legitimately holds a message some of them have
   * already sent: what is pending is the forward, not the message.
   * <p>
   * It is kept per domain because a domain sends this message whenever its
   * replication service is disabled - an online import, a restore, a
   * configuration change - and not only when the process shuts down. A single
@@ -63,6 +69,10 @@
   * It is kept per replica because the collocated RS relays the message of
   * every replica connected to it, and the forward of another replica's
   * message says nothing about this one.
   * <p>
   * Each entry knows the replication servers its message was queued for, because each of them is
   * served by its own writer: the forward of one of them says nothing about the others, whose
   * queue the shutdown is about to clear.
   */
  private final ConcurrentMap<DN, ConcurrentMap<Integer, PendingOfflineMsg>> replicaOfflineMsgs =
      new ConcurrentHashMap<>();
@@ -103,14 +113,54 @@
  }

  /**
   * Message has been forwarded.
   * Message has been queued for the replication servers which must forward it.
   * <p>
   * This must be called before the message is queued for any of them: a replication server can
   * forward it as soon as it is in its queue, and a forward which finds no recipient recorded
   * ends the wait at once.
   *
   * @param baseDN
   *          the domain for which the message has been sent
   * @param offlineCSN
   *          the CSN of the message which is being queued
   * @param replicationServerIds
   *          the server ids of the replication servers the message is being queued for
   */
  public void replicaOfflineMsgDispatched(
      DN baseDN, CSN offlineCSN, Collection<Integer> replicationServerIds)
  {
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs == null)
    {
      return;
    }
    final int serverId = offlineCSN.getServerId();
    final PendingOfflineMsg pending = msgs.get(serverId);
    /*
     * The message being queued may be an older announcement of the same replica - one which was
     * queued behind a backlog since an earlier import. The replication servers it goes to say
     * nothing about the announcement the shutdown is waiting for.
     */
    if (pending != null && pending.csn.equals(offlineCSN)
        && pending.awaitForwardsFrom(replicationServerIds))
    {
      // queued for nobody: there is nothing to wait for
      msgs.remove(serverId, pending);
      notifyForwarded();
    }
  }

  /**
   * Message has been forwarded to one of the replication servers it was queued for.
   *
   * @param baseDN
   *          the domain for which the message has been sent
   * @param forwardedCSN
   *          the CSN of the forwarded message
   * @param replicationServerId
   *          the server id of the replication server the message has been forwarded to
   */
  public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN)
  public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN, int replicationServerId)
  {
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs != null)
@@ -124,11 +174,47 @@
       * Such a forward says nothing about the announcement the shutdown is waiting for, and must
       * not consume its grace period.
       */
      if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN))
      if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN)
          && pending.forwardedBy(replicationServerId))
      {
        msgs.remove(serverId, pending);
      }
    }
    notifyForwarded();
  }

  /**
   * A replication server the message may have been queued for will not forward it: it is gone,
   * or the message was dropped on its way out.
   * <p>
   * Whatever it was given can no longer reach it, so the shutdown must not spend the rest of its
   * grace period waiting for it.
   *
   * @param baseDN
   *          the domain the replication server is connected to
   * @param replicationServerId
   *          the server id of the replication server which will not forward the message
   */
  public void replicaOfflineMsgNotForwarded(DN baseDN, int replicationServerId)
  {
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs != null)
    {
      for (Entry<Integer, PendingOfflineMsg> entry : msgs.entrySet())
      {
        final PendingOfflineMsg pending = entry.getValue();
        if (pending.giveUpOn(replicationServerId))
        {
          msgs.remove(entry.getKey(), pending);
        }
      }
    }
    notifyForwarded();
  }

  /** Wakes up the shutdown, which re-reads what is left to wait for. */
  private void notifyForwarded()
  {
    synchronized (forwardedMonitor)
    {
      forwardedMonitor.notifyAll();
@@ -137,7 +223,8 @@

  /**
   * Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg
   * has been forwarded or its grace period has expired.
   * has been forwarded by every replication server it was queued for, or its
   * grace period has expired.
   * <p>
   * The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)}
   * rather than polling this; it is the same state, observable without waiting for it.
@@ -269,6 +356,11 @@
    private final CSN csn;
    /** When the message was announced, on the {@link System#nanoTime()} clock. */
    private final long sentTime;
    /**
     * The replication servers the message was queued for and which have not forwarded it yet,
     * null as long as it has not been queued for anybody.
     */
    private volatile Set<Integer> awaitedForwarders;

    private PendingOfflineMsg(CSN csn, long sentTime)
    {
@@ -276,10 +368,56 @@
      this.sentTime = sentTime;
    }

    /**
     * Records the replication servers the message is being queued for, and returns whether there
     * is none of them, i.e. nothing left to wait for.
     */
    private boolean awaitForwardsFrom(Collection<Integer> replicationServerIds)
    {
      final Set<Integer> awaited = ConcurrentHashMap.newKeySet();
      awaited.addAll(replicationServerIds);
      awaitedForwarders = awaited;
      return awaited.isEmpty();
    }

    /**
     * Records the forward of one replication server, and returns whether nothing is left to wait
     * for.
     */
    private boolean forwardedBy(int replicationServerId)
    {
      final Set<Integer> awaited = awaitedForwarders;
      if (awaited == null)
      {
        /*
         * The message never went through the collocated RS - a replica which picked a remote one
         * announcing itself offline, or an announcement recorded after the message it belongs to
         * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait
         * had before the recipients were tracked: the first forward ends it.
         */
        return true;
      }
      awaited.remove(replicationServerId);
      return awaited.isEmpty();
    }

    /**
     * Gives up on the forward of one replication server, and returns whether nothing is left to
     * wait for. Unlike a forward, this releases nothing while no recipient is known: a peer going
     * away says nothing about a message it was never given.
     */
    private boolean giveUpOn(int replicationServerId)
    {
      final Set<Integer> awaited = awaitedForwarders;
      return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
    }

    @Override
    public String toString()
    {
      return "PendingOfflineMsg(" + csn + ")";
      final Set<Integer> awaited = awaitedForwarders;
      return "PendingOfflineMsg(" + csn
          + (awaited != null ? ", awaiting the forward of " + awaited : "") + ")";
    }
  }
}

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

@@ -19,30 +19,39 @@
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collection;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
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.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.DeleteMsg;
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.Session;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.protocol.WindowMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.StaticUtils;
@@ -57,8 +66,9 @@
 * <p>
 * Most tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory
 * server: the contract they pin is when the shutdown of the replication server waits, and how
 * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} pins the outcome
 * those waits exist for, on a peer connected through the real handshake.
 * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} and
 * {@link #theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline()} pin the outcome those
 * waits exist for, on peers connected through the real handshake.
 */
@SuppressWarnings("javadoc")
public class ReplicationServerShutdownSyncTest extends ReplicationTestCase
@@ -70,8 +80,25 @@
  private static final int REMOTE_DS_ID = 93;
  /** The collocated replica whose ReplicaOfflineMsg the shutdown waits for. */
  private static final int LOCAL_DS_ID = 94;
  /** The peer replication server whose writer is held back by a full send window. */
  private static final int HELD_BACK_RS_ID = 95;
  /** The peer replication server whose handshake is aborted while the message is pushed. */
  private static final int ABORTED_RS_ID = 96;
  /** Send window a peer advertises when nothing has to hold its writer back. */
  private static final int PEER_WINDOW = 100;
  /**
   * Send window of the peer which is held back: one change fills it, and the message which
   * follows stays with its writer until the peer gives it credit again.
   */
  private static final int HELD_BACK_PEER_WINDOW = 1;
  /** Time given to the forwarding thread before it releases the shutdown. */
  private static final long FORWARD_DELAY = 500;
  /**
   * Time given to a writer to reach the message it was handed once its send window is opened,
   * well short of the grace period so that a writer which never gets there is reported as such
   * rather than as a shutdown which waited.
   */
  private static final long WRITER_REACTION_TIMEOUT_MS = 10000;
  /** How often the domains of {@link #theGracePeriodIsSharedByAllTheDomainsOfOneShutdown()}
   * announce themselves offline again while the shutdown is waiting for them. */
  private static final long REANNOUNCE_INTERVAL = 200;
@@ -170,8 +197,8 @@

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain);
      final Future<ReplicaOfflineMsg> received = peer.receiveReplicaOfflineMsg();
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
      final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);

      /*
       * The replica announces itself offline once the shutdown of the replication server is
@@ -189,7 +216,7 @@
      final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      assertThat(forwarded)
          .as("the peer replication server was never told that the replica went offline, its "
              + "read ended with: %s", peer.readerFailure())
              + "read ended with: %s", peer.failure())
          .isNotNull();
      assertThat(forwarded.getCSN().getServerId()).isEqualTo(LOCAL_DS_ID);
      assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
@@ -266,6 +293,389 @@
  }

  /**
   * The grace period covers every peer replication server, not only the fastest of them. The
   * message is queued for each of them and published by its own writer, and the shutdown clears
   * the queue and closes the session of whoever has not published it yet: a peer whose writer is
   * held back would be left unaware that the replica went offline, and its change number indexer
   * would keep the medium consistency point pinned to the last change of that replica.
   */
  @Test
  public void theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    FakePeerReplicationServer heldBackPeer = null;
    Thread windowOpener = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncEveryPeerDb", 8229, replicationPort);
      broker =
          openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(
          replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW);
      heldBackPeer = new FakePeerReplicationServer(
          replicationPort, HELD_BACK_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
      waitForConnectedReplicationServer(domain, HELD_BACK_RS_ID);

      /*
       * One change fills the send window of the held back peer: its writer publishes that one and
       * then blocks on the permit of the next message, so the ReplicaOfflineMsg stays with it
       * while its neighbour forwards the same message right away.
       */
      final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
      final Future<DeleteMsg> windowFiller = heldBackPeer.receive(DeleteMsg.class);
      broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
          csns.newCSN(), "offline-entry-uuid"));
      assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the send window of the held back peer was never filled, its exchange ended "
              + "with: %s", heldBackPeer.failure())
          .isNotNull();

      final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
      final Future<ReplicaOfflineMsg> receivedWhenHeldBack =
          heldBackPeer.receive(ReplicaOfflineMsg.class);
      final CSN offlineCSN = csns.newCSN();
      shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
      broker.publish(new ReplicaOfflineMsg(offlineCSN));
      windowOpener = newWindowOpenerThread(heldBackPeer);

      final long startTime = System.nanoTime();
      windowOpener.start();
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      /*
       * The barrier first, because it is the one the shutdown is made of and it cannot race:
       * the wait ends when both writers have reported, or when the grace period runs out, and
       * the duration below tells the two apart.
       */
      assertThat(shutdownSync.forwardedBy())
          .as("the wait ended on the first peer forwarding the message, without the writer of "
              + "the peer which was held back ever reporting one")
          .contains(REMOTE_RS_ID, HELD_BACK_RS_ID);
      assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
          .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the peer which was not held back never received the message, its read ended "
              + "with: %s", peer.failure())
          .isNotNull();
      /*
       * The forward asserted above proves the message reached the Session, not the wire: close()
       * discards whatever is still in its send queue without draining it, which is the
       * limitation issue #919 recorded. If this is the only assertion which fails, that window
       * is the explanation rather than the granularity of the barrier.
       */
      assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the peer which was held back never learned that the replica went offline, "
              + "although its writer reported the forward, its exchange ended with: %s",
              heldBackPeer.failure())
          .isNotNull();
    }
    finally
    {
      joinQuietly(windowOpener);
      closeQuietly(heldBackPeer);
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * A message its writer drops on the way out will never be forwarded, so the shutdown must stop
   * waiting for the peer it was queued for. The filter of the writer is wider than the one
   * ReplicationServerDomain.put() applies when it queues the message - it drops anything for a
   * peer whose generation id is unknown as well - so a peer can be given a message which is then
   * dropped, and nothing would ever report a forward for it.
   */
  @Test
  public void theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncDroppedMsgDb", 8230, replicationPort);
      broker =
          openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(
          replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
      final ReplicationServerHandler rsHandler = domain.getConnectedRSs().get(REMOTE_RS_ID);

      /*
       * One change fills the send window of the peer, so the message which follows stays with
       * its writer until the window is opened again. The writer takes the message off the queue
       * before it blocks on the permit and evaluates its filter only once it has it, which is
       * what makes the generation id below take effect on a message already queued.
       */
      final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
      final Future<DeleteMsg> windowFiller = peer.receive(DeleteMsg.class);
      broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
          csns.newCSN(), "offline-entry-uuid"));
      assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the send window of the peer was never filled, its exchange ended with: %s",
              peer.failure())
          .isNotNull();

      final CSN offlineCSN = csns.newCSN();
      shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
      broker.publish(new ReplicaOfflineMsg(offlineCSN));
      shutdownSync.awaitDispatch();
      assertThat(shutdownSync.dispatchedTo())
          .as("the message was never queued for the peer, so its writer has nothing to drop")
          .contains(REMOTE_RS_ID);

      // the peer no longer shares the generation id of the domain, so its writer drops what was
      // queued for it before that - a filter wider than the one put() applied
      rsHandler.setGenerationId(domain.getGenerationId() + 1);
      peer.openSendWindow(PEER_WINDOW);
      awaitGiveUpOn(shutdownSync, REMOTE_RS_ID,
          "the writer dropped the message without telling the shutdown to stop waiting for the "
              + "peer it was queued for");

      final long startTime = System.nanoTime();
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      assertThat(elapsed)
          .as("the shutdown kept waiting for a forward its own writer had already dropped")
          .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
    }
    finally
    {
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * A peer whose handshake is aborted while the message is being pushed must not be waited for.
   * <p>
   * put() reads the peers of the domain, records them as the recipients of the message and only
   * then queues it for each of them. unregisterFailedHandshake() runs on the handshake thread
   * and takes no domain lock, so it can land in between: its own give-up then finds nothing
   * recorded yet and does nothing. Nothing else would ever strike that peer off - the reader and
   * writer threads whose death reaches stopServer() were never started for a handshake which was
   * aborted - so the shutdown waits out its whole grace period for a forward which cannot come.
   */
  @Test
  public void theShutdownStopsWaitingForAPeerWhoseHandshakeWasAborted() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = newReplicationServer(
          shutdownSync, "shutdownSyncAbortedHandshakeDb", 8231, replicationPort);
      broker =
          openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);

      final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        // a second peer, registered as a handshake does just before it fails
        final ReplicationServerHandler aborting = registerConnectedReplicationServer(
            replicationServer, baseDN, session, ABORTED_RS_ID);
        aborting.setGenerationId(domain.getGenerationId());
        shutdownSync.runWhileDispatching(new Runnable()
        {
          @Override
          public void run()
          {
            domain.unregisterFailedHandshake(aborting);
          }
        });

        final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
        final CSN offlineCSN = newOfflineCSN();
        shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
        broker.publish(new ReplicaOfflineMsg(offlineCSN));
        shutdownSync.awaitDispatch();

        final long startTime = System.nanoTime();
        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(shutdownSync.dispatchedTo())
            .as("the peer whose handshake was aborted was not recorded among the recipients, so "
                + "this test never reproduced the window it is about")
            .contains(REMOTE_RS_ID, ABORTED_RS_ID);
        assertThat(elapsed)
            .as("the shutdown waited for a peer whose handshake had been aborted, and which "
                + "nothing else will ever strike off")
            .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
        assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
            .as("the peer which was still connected never learned that the replica went "
                + "offline, its read ended with: %s", peer.failure())
            .isNotNull();
      }
    }
    finally
    {
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * A peer which disconnects during the grace period can no longer forward what it was given -
   * its session is closed under its writer - so stopServer() must strike it off rather than let
   * the shutdown wait out the rest of its window for a peer which is already gone.
   */
  @Test
  public void theShutdownStopsWaitingForAPeerWhichDisconnected() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    FakePeerReplicationServer heldBackPeer = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = newReplicationServer(
          shutdownSync, "shutdownSyncDisconnectedPeerDb", 8232, replicationPort);
      broker =
          openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(
          replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW);
      heldBackPeer = new FakePeerReplicationServer(
          replicationPort, HELD_BACK_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
      waitForConnectedReplicationServer(domain, HELD_BACK_RS_ID);

      // the peer which disconnects is held back by a full send window, so that it cannot have
      // forwarded the message before it goes: what ends the wait must be the give-up
      final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
      final Future<DeleteMsg> windowFiller = heldBackPeer.receive(DeleteMsg.class);
      broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
          csns.newCSN(), "offline-entry-uuid"));
      assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the send window of the held back peer was never filled, its exchange ended "
              + "with: %s", heldBackPeer.failure())
          .isNotNull();

      final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
      final CSN offlineCSN = csns.newCSN();
      shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
      broker.publish(new ReplicaOfflineMsg(offlineCSN));
      shutdownSync.awaitDispatch();
      heldBackPeer.close();

      final long startTime = System.nanoTime();
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      assertThat(shutdownSync.dispatchedTo())
          .as("the message was not queued for the peer which then disconnected, so this test "
              + "never reproduced what it is about")
          .contains(REMOTE_RS_ID, HELD_BACK_RS_ID);
      assertThat(elapsed)
          .as("the shutdown kept waiting for a forward from a peer which had disconnected")
          .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the peer which stayed connected never received the message, its read ended "
              + "with: %s", peer.failure())
          .isNotNull();
    }
    finally
    {
      closeQuietly(heldBackPeer);
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * A peer which does not share the generation id of the domain is not given the message, so it
   * must not be recorded among the peers the shutdown waits for. The caller side check on
   * getConnectedRSs() does not cover this: such a peer is connected, and would make the domain
   * spend its grace period on a message it was never queued.
   */
  @Test
  public void theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = newReplicationServer(
          shutdownSync, "shutdownSyncOtherGenerationIdDb", 8233, replicationPort);
      broker =
          openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
      domain.getConnectedRSs().get(REMOTE_RS_ID).setGenerationId(domain.getGenerationId() + 1);

      final CSN offlineCSN = newOfflineCSN();
      shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
      broker.publish(new ReplicaOfflineMsg(offlineCSN));
      shutdownSync.awaitDispatch();

      final long startTime = System.nanoTime();
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      assertThat(shutdownSync.dispatchedTo())
          .as("the message was recorded as queued for a peer which does not share the "
              + "generation id of the domain, and which it was never queued for")
          .isEmpty();
      assertThat(elapsed)
          .as("the shutdown waited for a peer the message was not queued for")
          .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
    }
    finally
    {
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * With no other replication server connected there is nobody to forward the message to, so
   * waiting would only delay the shutdown of a standalone server by the whole grace period.
   */
@@ -311,6 +721,9 @@
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort);
      broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      // the writer this test is about exists only once the handshake of the directory server is
      // over, and the shutdown below would otherwise be free to abort that handshake instead
      waitForConnectedDirectoryServer(replicationServer.getReplicationServerDomain(baseDN, true));

      final long startTime = System.nanoTime();
      shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
@@ -434,13 +847,20 @@
   * protocol exchange: the handler this leaves behind has no writer, which is enough for the
   * tests which only need a domain with a connected peer.
   */
  private void registerConnectedReplicationServer(
  private ReplicationServerHandler registerConnectedReplicationServer(
      ReplicationServer replicationServer, DN baseDN, Session session) throws Exception
  {
    return registerConnectedReplicationServer(replicationServer, baseDN, session, REMOTE_RS_ID);
  }

  private ReplicationServerHandler registerConnectedReplicationServer(
      ReplicationServer replicationServer, DN baseDN, Session session, int serverId)
      throws Exception
  {
    final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true);
    final ReplicationServerHandler rsHandler =
        new ReplicationServerHandler(session, 100, replicationServer, 100);
    rsHandler.serverId = REMOTE_RS_ID;
    rsHandler.serverId = serverId;
    rsHandler.serverURL = "127.0.0.1:1636";
    rsHandler.setBaseDNAndDomain(baseDN, false);
    domain.lock();
@@ -452,22 +872,68 @@
    {
      domain.release();
    }
    return rsHandler;
  }

  private void waitForConnectedReplicationServer(final ReplicationServerDomain domain)
      throws Exception
  /** Waits for the barrier to have been told that this peer will not forward the message. */
  private void awaitGiveUpOn(final RecordingShutdownSync shutdownSync, final int serverId,
      final String reason) throws Exception
  {
    new TestTimer.Builder()
        .maxSleep(WRITER_REACTION_TIMEOUT_MS, TimeUnit.MILLISECONDS)
        .sleepTimes(10, TimeUnit.MILLISECONDS)
        .toTimer()
        .repeatUntilSuccess(new TestTimer.CallableVoid()
        {
          @Override
          public void call() throws Exception
          {
            assertThat(shutdownSync.gaveUpOn()).as(reason).contains(serverId);
          }
        });
  }

  /**
   * Waits for the peer replication server to be connected <em>and</em> for its handshake to be
   * over.
   * <p>
   * The registration is not the end of the handshake: {@code startFromRemoteRS()} puts the
   * handler in {@code connectedRSs} before it calls {@code finalizeStart()}, which is what
   * starts the reader and the writer, and the whole handshake runs in the listen thread of the
   * replication server. {@link ReplicationServer#shutdown()} interrupts that thread before it
   * waits for the ReplicaOfflineMsgs, so a shutdown triggered while the handshake is still in
   * {@code Session.waitForStartup()} aborts it: the session is closed and the handler
   * unregistered, and the peer is gone before the message could be queued for it, let alone
   * forwarded. Issue #821 recorded that same window, from the dead handler it used to leave
   * behind.
   * <p>
   * The listen thread serves one handshake at a time, so a peer which is past that window also
   * puts every connection accepted before it - the collocated directory server of these tests
   * among them - past it.
   */
  private void waitForConnectedReplicationServer(
      final ReplicationServerDomain domain, final int serverId) throws Exception
  {
    newConnectionTimer().repeatUntilSuccess(new TestTimer.CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        assertThat(domain.getConnectedRSs())
            .as("the peer replication server never connected").containsKey(REMOTE_RS_ID);
        final ReplicationServerHandler rsHandler = domain.getConnectedRSs().get(serverId);
        assertThat(rsHandler)
            .as("the peer replication server %s never connected", serverId).isNotNull();
        assertThat(handshakeIsOver(rsHandler))
            .as("the handshake of the peer replication server %s never finished", serverId)
            .isTrue();
      }
    });
  }

  /**
   * Waits for the collocated directory server to be connected and for its handshake to be over -
   * {@link #waitForConnectedReplicationServer(ReplicationServerDomain, int)} says what the
   * registration alone leaves open.
   */
  private DataServerHandler waitForConnectedDirectoryServer(final ReplicationServerDomain domain)
      throws Exception
  {
@@ -478,11 +944,24 @@
      {
        final DataServerHandler dsHandler = domain.getConnectedDSs().get(REMOTE_DS_ID);
        assertThat(dsHandler).as("the directory server never connected").isNotNull();
        assertThat(handshakeIsOver(dsHandler))
            .as("the handshake of the directory server never finished").isTrue();
        return dsHandler;
      }
    });
  }

  /**
   * Whether the handshake of the handler is over, so that the interrupt
   * {@link ReplicationServer#shutdown()} sends to its listen thread can no longer abort it:
   * {@code ServerHandler.finalizeStart()} registers the handler as a monitor provider by its
   * last statement, after the reader and the writer have been started.
   */
  private static boolean handshakeIsOver(ServerHandler handler)
  {
    return DirectoryServer.getMonitorProviders().containsValue(handler);
  }

  private static TestTimer newConnectionTimer()
  {
    return new TestTimer.Builder()
@@ -503,7 +982,7 @@
        {
          return;
        }
        shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN);
        shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN, REMOTE_RS_ID);
      }
    });
  }
@@ -524,6 +1003,26 @@
    });
  }

  /**
   * Gives the held back peer credit to receive again, once the shutdown has had time to end on
   * the forward of the peer which was not held back.
   */
  private Thread newWindowOpenerThread(final FakePeerReplicationServer heldBackPeer)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        if (!sleepQuietly(FORWARD_DELAY))
        {
          return;
        }
        heldBackPeer.openSendWindow(PEER_WINDOW);
      }
    });
  }

  private Thread newReAnnouncerThread(final DSRSShutdownSync shutdownSync, final DN baseDN1,
      final DN baseDN2, final AtomicBoolean stopped)
  {
@@ -691,6 +1190,85 @@
  }

  /**
   * A synchronization object which records what the production code reports to it, so that a
   * test can assert on the barrier the shutdown is made of rather than only on how long it took,
   * and can interleave a teardown with the dispatch of a message.
   */
  private static final class RecordingShutdownSync extends DSRSShutdownSync
  {
    private final List<Integer> dispatchedTo = new CopyOnWriteArrayList<>();
    private final List<Integer> forwardedBy = new CopyOnWriteArrayList<>();
    private final List<Integer> gaveUpOn = new CopyOnWriteArrayList<>();
    private final CountDownLatch dispatched = new CountDownLatch(1);
    /** Runs inside the next dispatch, before the recipients are recorded. */
    private final AtomicReference<Runnable> whileDispatching = new AtomicReference<>();

    /**
     * Runs the provided action inside the next dispatch, before the recipients are recorded:
     * the window in which put() has read the peers of the domain and nothing knows yet which of
     * them the message is for.
     */
    void runWhileDispatching(Runnable action)
    {
      whileDispatching.set(action);
    }

    @Override
    public void replicaOfflineMsgDispatched(
        DN baseDN, CSN offlineCSN, Collection<Integer> replicationServerIds)
    {
      final Runnable action = whileDispatching.getAndSet(null);
      if (action != null)
      {
        action.run();
      }
      dispatchedTo.addAll(replicationServerIds);
      super.replicaOfflineMsgDispatched(baseDN, offlineCSN, replicationServerIds);
      dispatched.countDown();
    }

    @Override
    public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN, int replicationServerId)
    {
      forwardedBy.add(replicationServerId);
      super.replicaOfflineMsgForwarded(baseDN, forwardedCSN, replicationServerId);
    }

    @Override
    public void replicaOfflineMsgNotForwarded(DN baseDN, int replicationServerId)
    {
      gaveUpOn.add(replicationServerId);
      super.replicaOfflineMsgNotForwarded(baseDN, replicationServerId);
    }

    /** Waits for put() to have pushed a ReplicaOfflineMsg to the peers of its domain. */
    void awaitDispatch() throws InterruptedException
    {
      assertThat(dispatched.await(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("no ReplicaOfflineMsg was ever pushed to the peers of the domain")
          .isTrue();
    }

    /** The peers the message was recorded as queued for. */
    List<Integer> dispatchedTo()
    {
      return dispatchedTo;
    }

    /** The peers whose writer reported having forwarded the message. */
    List<Integer> forwardedBy()
    {
      return forwardedBy;
    }

    /** The peers the shutdown was told to stop waiting for. */
    List<Integer> gaveUpOn()
    {
      return gaveUpOn;
    }
  }

  /**
   * A peer replication server which connects to the replication server under test and completes
   * the handshake, so that the handler it leaves behind on the domain has a real writer and can
   * actually forward what the domain pushes to it.
@@ -699,12 +1277,23 @@
  {
    private final Session session;
    private final ExecutorService reader = Executors.newSingleThreadExecutor();
    /** Why the peer stopped reading, so that a missing message can be told from a failed one. */
    private volatile Exception readerFailure;
    /**
     * What ended the exchange with the replication server, so that a message which never arrived
     * can be told from an exchange which failed. The reader and the thread which opens the send
     * window both report here, and the first failure is the one kept: it is the one which
     * explains the rest.
     */
    private final AtomicReference<Exception> failure = new AtomicReference<>();

    FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId)
        throws Exception
    {
      this(replicationPort, serverId, baseDN, generationId, PEER_WINDOW);
    }

    FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId,
        int windowSize) throws Exception
    {
      final Socket socket = new Socket();
      Session newSession = null;
      boolean handshaken = false;
@@ -716,7 +1305,7 @@

        final String serverURL = "127.0.0.1:" + socket.getLocalPort();
        final byte groupId = (byte) 1;
        newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, 100,
        newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, windowSize,
            new ServerState(), generationId, false, groupId, 5000));
        final ReplServerStartMsg inStartMsg =
            waitForSpecificMsg(newSession, ReplServerStartMsg.class);
@@ -748,22 +1337,25 @@
      session = newSession;
    }

    /** Returns the first ReplicaOfflineMsg this peer receives, or null if its session ends first. */
    Future<ReplicaOfflineMsg> receiveReplicaOfflineMsg()
    /**
     * Returns the first message of the given type this peer receives, or null if its session
     * ends first.
     */
    <T extends ReplicationMsg> Future<T> receive(final Class<T> msgClass)
    {
      return reader.submit(new Callable<ReplicaOfflineMsg>()
      return reader.submit(new Callable<T>()
      {
        @Override
        public ReplicaOfflineMsg call()
        public T call()
        {
          try
          {
            while (true)
            {
              final ReplicationMsg msg = session.receive();
              if (msg instanceof ReplicaOfflineMsg)
              if (msgClass.isInstance(msg))
              {
                return (ReplicaOfflineMsg) msg;
                return msgClass.cast(msg);
              }
            }
          }
@@ -771,17 +1363,35 @@
          {
            // The session is closed when the replication server completes its shutdown: whatever
            // has not arrived by then never will.
            readerFailure = e;
            failed(e);
            return null;
          }
        }
      });
    }

    /** Returns what ended the read of this peer, null if nothing did. */
    Exception readerFailure()
    /** Gives the replication server credit to publish again, as a peer which keeps up does. */
    void openSendWindow(int credits)
    {
      return readerFailure;
      try
      {
        session.publish(new WindowMsg(credits));
      }
      catch (IOException e)
      {
        failed(e);
      }
    }

    private void failed(Exception e)
    {
      failure.compareAndSet(null, e);
    }

    /** Returns what ended the exchange with this peer, null if nothing did. */
    Exception failure()
    {
      return failure.get();
    }

    void close()

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java

@@ -18,6 +18,7 @@
import static java.util.Arrays.asList;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

import org.forgerock.opendj.ldap.DN;
@@ -41,6 +42,9 @@
  private static final long FORWARD_DELAY = 200;
  private static final int SERVER_ID = 1;
  private static final int OTHER_SERVER_ID = 2;
  /** A peer replication server the collocated one relays the message to. */
  private static final int RS_ID = 11;
  private static final int OTHER_RS_ID = 12;

  private static DN baseDN1;
  private static DN baseDN2;
@@ -77,7 +81,7 @@
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }
@@ -106,7 +110,7 @@
    // an import disables then re-enables the replication service
    final CSN sentByTheImport = newCSN(SERVER_ID, 1);
    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheImport);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport, RS_ID);
    Thread.sleep(GRACE_PERIOD + 50);

    // the shutdown of the process, much later
@@ -129,11 +133,11 @@

    shutdownSync.replicaOfflineMsgSent(baseDN1, queuedByAnEarlierImport);
    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();

    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }
@@ -161,7 +165,7 @@
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID), RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }
@@ -175,7 +179,7 @@

    shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(OTHER_SERVER_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }
@@ -189,8 +193,8 @@

    shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheOtherReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica, RS_ID);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }
@@ -247,6 +251,206 @@
        .isLessThan(2 * GRACE_PERIOD);
  }

  /**
   * The collocated replication server queues the message for every peer it relays to, and each
   * of them is served by its own writer: the forward of one peer says nothing about the others,
   * whose queue the shutdown is about to clear.
   */
  @Test
  public void theForwardOfOnePeerDoesNotEndTheWaitOfTheOthers() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  @Test
  public void canShutdownOnceEveryPeerTheMessageWasQueuedForForwardedIt() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, OTHER_RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * With no peer to relay the message to - none connected, or none sharing the generation id of
   * the domain - there is nothing to wait for.
   */
  @Test
  public void canShutdownWhenTheMessageWasQueuedForNoPeer() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(
        baseDN1, offlineCSN, Collections.<Integer> emptyList());

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * A peer which is no longer connected cannot forward anything, so the shutdown must not spend
   * the rest of its window waiting for it.
   */
  @Test
  public void aPeerWhichStoppedIsNoLongerWaitedFor() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
    shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, OTHER_RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /** The peers which are still connected keep their part of the grace period. */
  @Test
  public void aPeerWhichStoppedDoesNotEndTheWaitOfTheOthers() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, OTHER_RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  /**
   * A peer which connected after the message was queued was never given it, so what it forwards
   * is a message of its own catch-up and says nothing about the peers which still owe theirs.
   */
  @Test
  public void theForwardOfAPeerTheMessageWasNotQueuedForDoesNotEndTheWait() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, OTHER_RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  /**
   * The peers are recorded by the collocated replication server when it queues the message for
   * them, which the message of a replica connected to a remote replication server never reaches.
   * With no peer recorded the wait keeps the behaviour it had before they were tracked: the
   * first forward ends it.
   */
  @Test
  public void theFirstForwardEndsTheWaitWhenNoPeerWasRecorded() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * A peer going away says nothing about a message it was never given, which is the opposite of
   * what a forward says: with no peer recorded the first forward ends the wait, and a give-up
   * must leave it running. Otherwise any peer disconnecting would release a message the
   * collocated replication server has not queued for anybody yet - the very bug the recipients
   * were introduced to close, in a new shape.
   */
  @Test
  public void aPeerStoppingBeforeTheMessageIsQueuedDoesNotEndTheWait() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  /**
   * A replica announces itself offline on every disableService(), so the peers recorded for an
   * earlier announcement say nothing about the one the shutdown is waiting for.
   */
  @Test
  public void thePeersOfAnEarlierAnnouncementAreNotTakenForThoseOfThisOne() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN queuedByAnEarlierImport = newCSN(SERVER_ID, 1);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgDispatched(
        baseDN1, queuedByAnEarlierImport, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * The peer which goes away must wake the shutdown up, and not leave it waiting for a forward
   * nobody can report any more.
   */
  @Test
  public void theWaitEndsWhenTheLastPeerExpectedToForwardStops() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);
    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID));
    final Thread peerStopper = newPeerStopperThread(shutdownSync, RS_ID);

    final long startTime = System.nanoTime();
    peerStopper.start();
    shutdownSync.awaitReplicaOfflineMsgsForwarded(
        asList(baseDN1), shutdownSync.newShutdownDeadline());
    final long elapsed = millisSince(startTime);
    peerStopper.join();

    assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY);
    assertThat(elapsed)
        .as("the peer going away did not wake the wait up")
        .isLessThan(LONG_GRACE_PERIOD);
  }

  /** Stops the peer the message was queued for, as a disconnection during the wait does. */
  private Thread newPeerStopperThread(final DSRSShutdownSync shutdownSync, final int peerId)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, peerId);
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
        }
      }
    });
  }

  /** Forwards the message of the first domain, then, as long again later, of the second one. */
  private Thread newForwarderThread(final DSRSShutdownSync shutdownSync,
      final CSN ofTheFirstDomain, final CSN ofTheSecondDomain)
@@ -259,9 +463,9 @@
        try
        {
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain);
          shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain, RS_ID);
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain);
          shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain, RS_ID);
        }
        catch (InterruptedException e)
        {