[#909] Cover the change a stopped replay thread hands back to the replicati...

Valery Kharseko

9 hours ago d0422c684b5fc32280ec28813d2ab06a047ee63a

[#909] Cover the change a stopped replay thread hands back to the replication server (#941)

 opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java

@@ -23,10 +23,14 @@

import java.io.IOException;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -230,6 +234,24 @@

  /** {@inheritDoc} */
  @Override
  public void finalizePlugin()
  {
    /*
     * A park which outlives the test which took it holds a replay thread of this server,
     * and every replayed operation queued behind it, for as long as this plugin is
     * loaded: the map is static and nothing but the test itself removes an entry from it.
     */
    for (ParkedReplay park : parks.values())
    {
      park.deregister();
    }
    parks.clear();
  }



  /** {@inheritDoc} */
  @Override
  public PluginResult.PreParse
         doPreParse(PreParseAbandonOperation abandonOperation)
  {
@@ -616,7 +638,7 @@
    }

    // Check for registered short circuits.
    final String key = operation.getOperationType() + "/" + section.toLowerCase();
    final String key = keyFor(operation.getOperationType(), section);
    Integer resultCode = shortCircuits.get(key);
    if (resultCode != null)
    {
@@ -630,6 +652,26 @@
      // operations are let through, which is how a transient failure is simulated.
    }

    /*
     * A parked replay is held here, which is inside the run() of the operation and before
     * anything of the backend was taken: the thread which is replaying a change sits on
     * this monitor while it still owns that change, which is what lets a test act on the
     * thread rather than race it. It is consulted last, so that a park never takes an
     * operation away from a control or from a registered short circuit.
     */
    if (operation.isSynchronizationOperation())
    {
      final ParkedReplay park = parks.get(key);
      if (park != null && park.parks(operation))
      {
        final int parkResultCode = park.hold();
        if (parkResultCode >= 0)
        {
          return parkResultCode;
        }
      }
    }

    // If we've gotten here, then we shouldn't short-circuit the operation
    // processing.
    return -1;
@@ -688,7 +730,7 @@
   */
  public static int getShortCircuitCount(OperationType operation, String section)
  {
    final AtomicInteger count = shortCircuitCounts.get(operation + "/" + section.toLowerCase());
    final AtomicInteger count = shortCircuitCounts.get(keyFor(operation, section));
    return count != null ? count.get() : 0;
  }

@@ -701,7 +743,7 @@
   */
  public static void registerShortCircuit(OperationType operation, String section, int resultCode)
  {
    final String key = operation + "/" + section.toLowerCase();
    final String key = keyFor(operation, section);
    // This registration applies to every operation, and it counts from zero: a limit or
    // a count left behind by a previous registration is not part of it.
    shortCircuitCounts.remove(key);
@@ -720,7 +762,7 @@
   */
  public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes)
  {
    final String key = operation + "/" + section.toLowerCase();
    final String key = keyFor(operation, section);
    shortCircuitCounts.remove(key);
    shortCircuitLimits.put(key, maxTimes);
    shortCircuits.put(key, resultCode);
@@ -733,11 +775,310 @@
   */
  public static void deregisterShortCircuit(OperationType operation, String section)
  {
    final String key = operation + "/" + section.toLowerCase();
    final String key = keyFor(operation, section);
    shortCircuits.remove(key);
    shortCircuitLimits.remove(key);
    // The count belongs to the registration which is being removed: a test which counts
    // the operations it short circuits must not inherit the count of the previous one.
    shortCircuitCounts.remove(key);
  }

  /** Registered parks for the replayed operations, keyed like the short circuits. */
  private static final Map<String, ParkedReplay> parks = new ConcurrentHashMap<>();

  /**
   * Holds the replayed operations of one type where they are, one at a time, until the
   * test lets each of them go.
   * <p>
   * The hold is taken at a plugin point which runs inside {@code op.run()}, so the thread
   * which is replaying a change is stopped while it still owns that change: a test can
   * then do something to that thread - stop it, disable its domain - and know the change
   * is in flight rather than hope it is. Nothing of the backend has been taken at that
   * point, so a parked operation blocks the replay and nothing else.
   */
  public static final class ParkedReplay
  {
    /**
     * The value which lets the operation run rather than short circuit it.
     * <p>
     * {@code ResultCode.UNDEFINED} is registered on {@code -1} as well, so
     * {@code release(ResultCode.UNDEFINED.intValue())} lets the operation run instead of
     * making it report that code - the same hole {@code registerShortCircuit(-1)} has.
     * No caller has a use for it, and a park releases with a real result code or with
     * none at all.
     */
    private static final int LET_THROUGH = -1;

    /**
     * How long an operation is held before this park gives up on the test which took it.
     * <p>
     * It is far longer than any release a test waits for - the fixture itself waits a
     * minute for a park - and it exists for the test which never releases at all: a park
     * leaked by a method killed on a timeout would otherwise hold a replay thread of this
     * server, and every replayed operation queued behind it, for the life of the JVM.
     */
    private static final long MAX_HOLD_IN_MS = TimeUnit.MINUTES.toMillis(5);

    private final String key;
    /** Which of the replayed operations of that type this park is for. */
    private final Predicate<PluginOperation> parked;
    private final Object lock = new Object();
    /** Whether an operation is parked right now. */
    private boolean occupied;
    /**
     * The thread of the operation which parked last. It is never cleared, so that a test
     * which waited for a park is handed the thread of that park even when the operation
     * has left the park since - a park which is let go of by {@link #deregister()}, or by
     * the thread it holds being interrupted, would otherwise hand out no thread at all
     * and have an assertion on which thread replays the change pass without asserting it.
     */
    private Thread lastParkedThread;
    /** How many operations were parked, which is what tells one park from the next. */
    private int parkedOperations;
    /** How many of them the test has waited for already. */
    private int awaitedOperations;
    private boolean released;
    private int releasedResultCode;
    private boolean deregistered;

    private ParkedReplay(String key, Predicate<PluginOperation> parked)
    {
      this.key = key;
      this.parked = parked;
    }

    /** Returns whether the provided operation is one this park is for. */
    private boolean parks(PluginOperation operation)
    {
      return parked.test(operation);
    }

    /**
     * Parks the calling operation until the test releases it. Runs on the thread which is
     * replaying the change.
     *
     * @return the result code the operation must be short circuited with, or a negative
     *         value to let it run
     */
    private int hold()
    {
      final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(MAX_HOLD_IN_MS);
      synchronized (lock)
      {
        // One operation at a time, so that a release belongs to the operation the test
        // waited for rather than to whichever of them the scheduler let in first.
        while (occupied && !deregistered)
        {
          if (!waitOnLock(deadline))
          {
            return LET_THROUGH;
          }
        }
        if (deregistered)
        {
          return LET_THROUGH;
        }
        occupied = true;
        lastParkedThread = Thread.currentThread();
        parkedOperations++;
        released = false;
        lock.notifyAll();
        try
        {
          while (!released && !deregistered)
          {
            if (!waitOnLock(deadline))
            {
              return LET_THROUGH;
            }
          }
          return released ? releasedResultCode : LET_THROUGH;
        }
        finally
        {
          occupied = false;
          lock.notifyAll();
        }
      }
    }

    /**
     * Waits on the monitor until the provided deadline, reporting whether waiting can go
     * on. A deadline which has passed gives up on this park altogether rather than only
     * on the operation which reached it: the operations behind it would each pay the
     * whole wait again otherwise.
     */
    private boolean waitOnLock(long deadlineInNanos)
    {
      final long leftInNanos = deadlineInNanos - System.nanoTime();
      if (leftInNanos <= 0)
      {
        giveUpOnTheTest();
        return false;
      }
      try
      {
        // Rounded up, so that a budget shorter than a millisecond is still waited out
        // rather than truncated to a wait with no timeout at all.
        lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
        return true;
      }
      catch (InterruptedException e)
      {
        // Whatever wants this thread to stop wins over the park: let the operation run
        // rather than hold a thread which is being taken down.
        Thread.currentThread().interrupt();
        return false;
      }
    }

    /** Stops parking anything and says so, after a test held an operation for too long. */
    private void giveUpOnTheTest()
    {
      System.err.println("***** ERROR:  a replayed operation was parked on " + key
          + " for " + MAX_HOLD_IN_MS + " ms and was never released:  the test which took"
          + " this park left it behind.  Letting the operation run and parking no more.");
      deregister();
    }

    /**
     * Waits for a replayed operation which was not waited for yet to be parked, and
     * reports which thread is replaying it. The operations are parked one at a time, so
     * that thread is the one which was parked when this returns; the thread of the last
     * park is reported when several of them were let go of without being waited for.
     *
     * @param timeout how long to wait for it
     * @param unit the unit of the timeout
     * @return the thread which is replaying the parked operation
     * @throws InterruptedException if this thread is interrupted while waiting
     * @throws TimeoutException if no operation was parked in time
     * @throws IllegalStateException if this park is gone, so that nothing can be parked
     *           on it any more
     */
    public Thread awaitParked(long timeout, TimeUnit unit)
        throws InterruptedException, TimeoutException
    {
      final long deadline = System.nanoTime() + unit.toNanos(timeout);
      synchronized (lock)
      {
        while (parkedOperations <= awaitedOperations)
        {
          if (deregistered)
          {
            // Waiting out the budget here would report a timeout naming the operations
            // which never parked, rather than the park which cannot park them any more.
            throw new IllegalStateException("the park on " + key + " is gone - it was"
                + " deregistered, or displaced by another park of the same operations -"
                + " so no replayed operation will be parked on it again");
          }
          final long leftInNanos = deadline - System.nanoTime();
          if (leftInNanos <= 0)
          {
            throw new TimeoutException("no replayed operation was parked on " + key
                + " within " + timeout + " " + unit);
          }
          // Rounded up, so that a budget shorter than a millisecond is still waited out
          // rather than truncated to a wait with no timeout at all.
          lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
        }
        awaitedOperations = parkedOperations;
        return lastParkedThread;
      }
    }

    /**
     * Lets the parked operation run. Valid once {@link #awaitParked} has reported that
     * operation: see there for what a release which arrives before it costs.
     */
    public void release()
    {
      release(LET_THROUGH);
    }

    /**
     * Lets the parked operation go, short circuiting it with the provided result code.
     * <p>
     * Valid once {@link #awaitParked} has reported the operation being released. A
     * release which arrives before an operation is parked is wiped by the park it was
     * meant for - a park starts out unreleased - and that operation then waits for a
     * release which has already been spent.
     *
     * @param resultCode the result code the operation must report
     */
    public void release(int resultCode)
    {
      synchronized (lock)
      {
        if (!occupied)
        {
          throw new IllegalStateException("nothing is parked on " + key + " to release:"
              + " a release is spent by the park it arrives before, and the operation"
              + " which parks next then waits for one which has already been given");
        }
        released = true;
        releasedResultCode = resultCode;
        lock.notifyAll();
      }
    }

    /**
     * Stops parking the replayed operations and lets go of the one which is parked, if
     * any. A test must call this however it ends, or it leaves a replay thread of this
     * server parked for good.
     */
    public void deregister()
    {
      parks.remove(key, this);
      synchronized (lock)
      {
        deregistered = true;
        lock.notifyAll();
      }
    }
  }

  /**
   * Parks the replayed operations of the given type at the given plugin point, until the
   * test releases each of them.
   *
   * @param operation the type of operation to park
   * @param section the plugin point to park them at, which can only be {@code PreParse}
   * @param parked which of them to park - the change a test acts on rather than whatever
   *          of that type reaches this point first, which is somebody else's change as
   *          soon as more than one of them is in flight
   * @return the park, which the test must {@link ParkedReplay#deregister()} when it is
   *         done with it
   * @throws IllegalArgumentException if asked for any plugin point but {@code PreParse}
   */
  public static ParkedReplay parkReplayedOperations(
      OperationType operation, String section, Predicate<PluginOperation> parked)
  {
    if (!"PreParse".equalsIgnoreCase(section))
    {
      /*
       * The pre-operation plugins are not invoked for synchronization operations at all,
       * so a park anywhere else is never reached: the test which took it would wait out
       * its whole budget for an operation which cannot park, and be told that none did
       * rather than that none could.
       */
      throw new IllegalArgumentException("replayed operations can only be parked at"
          + " PreParse, which is the only plugin point they reach, not at " + section);
    }
    final String key = keyFor(operation, section);
    final ParkedReplay park = new ParkedReplay(key, parked);
    final ParkedReplay previous = parks.put(key, park);
    if (previous != null)
    {
      // A park a test left behind holds a replay thread of this server for good once the
      // map stops pointing at it: let go of it rather than lose the last reference to it.
      previous.deregister();
    }
    return park;
  }

  /** Returns the key a short circuit or a park of the given operations is kept under. */
  private static String keyFor(OperationType operation, String section)
  {
    return operation + "/" + section.toLowerCase(Locale.ROOT);
  }
}

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

@@ -102,6 +102,16 @@
  /** Generation id for a fully empty domain. */
  public static final long EMPTY_DN_GENID = GenerationIdChecksum.EMPTY_BACKEND_GENERATION_ID;

  /** The group a replication server and a replication domain are in unless told otherwise. */
  protected static final int DEFAULT_GROUP_ID = 1;

  /**
   * The group of a broker which is in none. Assured replication does not cross group ids,
   * so such a broker is never waited for and never waits: it is all a broker which only
   * publishes and reads updates needs.
   */
  private static final int NO_GROUP_ID = -1;

  /** How many times {@link #assertMonitorAttrValueStays} reads a value by default. */
  private static final int MONITOR_ATTR_SAMPLES = 5;

@@ -241,7 +251,42 @@
      int serverId, int windowSize, int port, int timeout,
      long generationId) throws Exception
  {
    final DomainFakeCfg config = newFakeCfg(baseDN, serverId, port);
    return openReplicationSession(
        newFakeCfg(baseDN, serverId, port), windowSize, timeout, generationId);
  }

  /**
   * Open a session to the local ReplicationServer which takes part in assured replication.
   * <p>
   * Assured replication does not cross group ids, so a broker whose updates are to be
   * acknowledged by the replicas of this server has to be in the group of the replication
   * server: an update published by a broker of another group is acknowledged on the spot,
   * by the replication server itself, and says nothing about what any replica did with it.
   * <p>
   * The group cuts both ways, and this broker does not acknowledge anything: the
   * replication server expects an ack from every replica of its group whatever that
   * replica is configured for, so a SAFE_READ update published by anyone else while this
   * broker is connected waits out the {@code assured-timeout} of the server. Publish the
   * assured updates from this broker, and open only one of them.
   *
   * @param baseDN the suffix the session is opened for
   * @param serverId the id this broker takes
   * @param windowSize the window size of the session
   * @param port the port of the local replication server
   * @param timeout the read timeout of the session, or 0 for none
   * @return the connected broker
   * @throws Exception if the session could not be opened
   */
  protected ReplicationBroker openAssuredReplicationSession(final DN baseDN,
      int serverId, int windowSize, int port, int timeout) throws Exception
  {
    return openReplicationSession(newFakeCfg(baseDN, serverId, port, DEFAULT_GROUP_ID),
        windowSize, timeout, getGenerationId(baseDN));
  }

  private ReplicationBroker openReplicationSession(final DomainFakeCfg config,
      int windowSize, int timeout, long generationId) throws Exception
  {
    config.setWindowSize(windowSize);

    final ReplicationBroker broker = new ReplicationBroker(
@@ -253,7 +298,13 @@

  protected DomainFakeCfg newFakeCfg(final DN baseDN, int serverId, int port)
  {
    DomainFakeCfg fakeCfg = new DomainFakeCfg(baseDN, serverId, newTreeSet("127.0.0.1:" + port));
    return newFakeCfg(baseDN, serverId, port, NO_GROUP_ID);
  }

  protected DomainFakeCfg newFakeCfg(final DN baseDN, int serverId, int port, int groupId)
  {
    DomainFakeCfg fakeCfg =
        new DomainFakeCfg(baseDN, serverId, newTreeSet("127.0.0.1:" + port), groupId);
    fakeCfg.setHeartbeatInterval(100000);
    fakeCfg.setChangetimeHeartbeatInterval(500);
    return fakeCfg;

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

@@ -22,6 +22,7 @@
import static org.forgerock.opendj.ldap.ModificationType.*;
import static org.forgerock.opendj.ldap.requests.Requests.*;
import static org.forgerock.opendj.ldap.schema.CoreSchema.*;
import static org.mockito.Mockito.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
@@ -32,6 +33,8 @@
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;

import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -44,6 +47,7 @@
import org.forgerock.opendj.ldap.requests.ModifyDNRequest;
import org.forgerock.opendj.ldap.requests.ModifyRequest;
import org.forgerock.opendj.ldap.schema.AttributeType;
import org.forgerock.opendj.server.config.server.ReplicationSynchronizationProviderCfg;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.AddOperation;
import org.opends.server.core.DeleteOperation;
@@ -52,11 +56,14 @@
import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.extensions.DummyAlertHandler;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.plugins.ShortCircuitPlugin.ParkedReplay;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.plugin.LDAPReplicationDomain;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.AckMsg;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.HeartbeatThread;
@@ -157,7 +164,15 @@
        + "cn: Replication Server\n"
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: UpdateOperationTest\n"
        + "ds-cfg-replication-server-id: 107\n";
        + "ds-cfg-replication-server-id: 107\n"
        /*
         * Long enough for a delivery which a test stops on its way through the replay:
         * the acks of an assured update are waited for from the moment it is published,
         * and the default second is spent long before a test which parks that delivery
         * has let go of it. Nothing waits it out - no test here leaves an assured update
         * unacknowledged - so it only bounds a failure.
         */
        + "ds-cfg-assured-timeout: 120000ms\n";

    // suffix synchronized
    String testName = "updateOperationTest";
@@ -1516,6 +1531,447 @@
  }

  /**
   * Test case for [Issue 909]: a replay thread which is stopped while it holds a change -
   * the number of replay threads is changed on a live server - must hand that change back
   * to the replication server instead of leaving it listed as owned by a thread which is
   * gone.
   * <p>
   * The change is parked inside the operation it is replayed by, so the thread is caught
   * while it still owns it rather than raced for: a change which is released by the
   * ordinary recovery instead ends the same way - delivered again and applied - so a test
   * which only watched the end state would pass whether or not the hand-back happened.
   * <p>
   * What tells them apart is which thread replays the change next. The thread which held
   * it is gone, and the change is replayed by one of the threads which replaced it, so
   * the delivery it is replayed from can only be a new one: an attempt which the same
   * thread made again would be the retry in place, and a change nobody handed back is
   * never delivered again at all - it stays listed as owned by a thread which is gone,
   * with the ServerState of this domain stopped behind it for good.
   */
  @Test
  public void aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain() throws Exception
  {
    testSetUp("aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain"));

    final int serverId = 14;
    /*
     * In the group of the replication server, so that the delete published below is one
     * this domain has to acknowledge: an assured update from a broker of another group is
     * acknowledged by the replication server itself, and says nothing about the replay.
     */
    ReplicationBroker broker =
        openAssuredReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
    try
    {
      CSNGenerator gen = new CSNGenerator(serverId, 0);

      Entry tmp = TestCaseUtils.addEntry(
          "dn: uid=user.909," + baseDN,
          "objectClass: top",
          "objectClass: person",
          "objectClass: organizationalPerson",
          "objectClass: inetOrgPerson",
          "uid: user.909",
          "cn: Aaccf Amar",
          "sn: Amar");
      String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();

      final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
      domain.resetUnreplayedChangeAlertThrottle();
      final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
      /*
       * Only the counters of the replay are read: a session restart takes this domain
       * through NOT_CONNECTED, which resets every monitoring counter of the replication
       * service - the updates it received and processed, and the assured acks it sent -
       * and handing a change back is a session restart.
       */
      final long initialApplied = getMonitorAttrValue(baseDN, "replayed-updates-ok");
      final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");

      /*
       * Hold the replayed deletes where they are. The park is taken at the pre-parse
       * plugin point, which runs inside op.run() and before anything of the backend was
       * taken: the replay thread stops there while it still owns the change, and the
       * pre-operation plugins are not invoked for synchronization operations at all.
       */
      final CSN csn = gen.newCSN();
      final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
          OperationType.DELETE, "PreParse", op -> csn.equals(OperationContext.getCSN(op)));
      final AtomicReference<Throwable> reconfigurationFailure = new AtomicReference<>();
      Thread reconfiguration = null;
      boolean reconfigurationFinished = true;
      try
      {
        final DeleteMsg delete = new DeleteMsg(tmp.getName(), csn, uuid);
        /*
         * Published assured in SAFE_READ mode, so that the delivery which is abandoned
         * has to say what it did. The ack is the one thing the counters cannot report
         * afterwards - the session restart the hand-back performs resets every one of
         * them - so it is read off this broker rather than counted.
         */
        delete.setAssured(true);
        delete.setAssuredMode(AssuredMode.SAFE_READ_MODE);
        broker.publish(delete);

        // A replay thread now owns the change and is stopped inside its operation.
        final Thread abandoningThread = parked.awaitParked(60, SECONDS);

        /*
         * Change the number of replay threads while that thread holds the change. It runs
         * on a thread of its own because stopping the replay threads joins them: it can
         * not return before the parked thread is let go, which is exactly the ordering
         * this test is about.
         */
        reconfiguration = startReplayThreadReconfiguration(2, reconfigurationFailure);
        awaitStoppingTheReplayThreads(reconfiguration, reconfigurationFailure);

        /*
         * The parked thread has been asked to stop by now, so its attempt comes back on a
         * storage which did not serve the operation - which is retried in place - and the
         * attempt which follows is where it finds out that it is going away and gives the
         * change back instead.
         */
        parked.release(ResultCode.UNAVAILABLE.intValue());

        /*
         * The change was given back, so the replication server owns it again and delivers
         * it once more. This second park is where the state of the domain is read: the
         * change is held inside its operation, so nothing can be recording it while the
         * assertions below run.
         */
        final Thread replayingThread =
            awaitParkedOrReportReconfigurationFailure(parked, reconfigurationFailure);
        reconfiguration.join(SECONDS.toMillis(60));
        assertFalse(reconfiguration.isAlive(),
            "the replay threads were reconfigured, but applyConfigurationChange never returned");
        if (reconfigurationFailure.get() != null)
        {
          throw new AssertionError("the replay threads could not be reconfigured",
              reconfigurationFailure.get());
        }

        /*
         * The change is being replayed by another thread, and the thread which held it is
         * gone: it gave the change back on its way out rather than take it with it. An
         * attempt made by the thread which held it would be the retry in place instead,
         * and a change which was never handed back is not replayed by anyone.
         */
        assertFalse(abandoningThread.isAlive(),
            "the replay thread which held the change must have stopped");
        Assertions.assertThat(replayingThread)
            .as("the change must be replayed by a thread which did not hold it")
            .isNotSameAs(abandoningThread);

        assertFalse(domain.getServerState().cover(csn),
            "a change a stopped replay thread never applied must not be in the ServerState");
        assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
            "a change which was handed back must not be counted as one this replica gave up on");
        assertNotNull(getEntry(tmp.getName(), 1, true),
            "the entry must not have been deleted by the delivery which was abandoned");

        /*
         * The delivery which was abandoned said what it did: an assured write which is
         * waiting on this replica must not be told that the change is in the data here,
         * because it is a change this replica is still asking for.
         *
         * What is read is the ack, not when it was published. The hand-back gives this
         * domain a new broker and the replication server waits for an ack by CSN rather
         * than by session, so an ack published after the hand-back reaches it all the
         * same: the order of the two is not what this asserts.
         */
        final AckMsg ack = awaitAck(broker, csn);
        assertTrue(ack.hasReplayError(),
            "the ack of the abandoned delivery must report the replay error rather than"
                + " be the plain ack a master would take for a durable write");
        assertFalse(ack.hasTimeout(),
            "the ack must be the one the abandoned delivery published, not the one the"
                + " replication server makes up when it gives up waiting for it");
        Assertions.assertThat(ack.getFailedServers())
            .as("the replica which abandoned the delivery must be the one it names")
            .containsExactly(domainSid);

        /*
         * Let the delivery which took over apply the change, and stop parking: an attempt
         * of that delivery which came back on a lock it could not take would be parked
         * again otherwise, with nothing left to release it.
         */
        parked.deregister();

        assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialApplied + 1,
            "the change must be applied by the delivery which took over from the abandoned one");
        /*
         * A change applied twice goes through the expected count on its way up, so the
         * value has to be seen to stay put rather than to be reached once - and for
         * longer than the session restart which would bring that second delivery, or the
         * assertion stops looking before what it is looking for could arrive.
         */
        assertMonitorAttrValueStays(baseDN, "replayed-updates-ok", initialApplied + 1,
            MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
            "the change must be applied exactly once");
        assertNull(DirectoryServer.getEntry(tmp.getName()), "the entry must have been deleted");

        /*
         * Abandoning a change is not giving up on it: the change is applied moments later,
         * so nothing is counted as failed and the administrator is not told that this
         * replica diverges.
         */
        assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
            MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
            "a change which was handed back must not be counted as one this replica gave up on");
        assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE),
            initialAlerts,
            "a change which was handed back must not have this replica report a divergence");
      }
      finally
      {
        /*
         * Whatever happened above: no replay thread of this server may be left parked,
         * and the reconfiguration has to be over before the next one is started. The two
         * would otherwise race for the pool of replay threads, which belongs to the
         * server rather than to this test, and the one which creates its threads last
         * drops the ones the other had just started without ever stopping them.
         */
        parked.deregister();
        if (reconfiguration != null)
        {
          reconfiguration.join(SECONDS.toMillis(60));
          reconfigurationFinished = !reconfiguration.isAlive();
        }
        if (reconfigurationFinished)
        {
          setNumberOfReplayThreads(null);
        }
      }
      /*
       * Read here rather than asserted while cleaning up, where a failure would replace
       * the one the test was reporting. join() with a timeout returns the same whether
       * the thread is over or not, and the pool of replay threads belongs to the server:
       * a reconfiguration still inside stopReplayThreads() holds the monitor of
       * MultimasterReplication, which restoring the pool would wait on for as long as an
       * untimed join() takes, and one caught between stopping and creating orphans the
       * threads the restore had just started. So the pool is left alone and this says so.
       */
      assertTrue(reconfigurationFinished,
          "the replay thread reconfiguration never finished; the pool was left alone");
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Starts changing the number of replay threads of this server on a thread of its own.
   * <p>
   * It cannot run on the thread of the test: stopping the replay threads joins them, so it
   * does not return while one of them is parked in a change, which is the whole point of
   * this fixture.
   *
   * @param replayThreads how many replay threads the server must run, or {@code null} for
   *          as many as it computes on its own
   * @param failure where the reconfiguration reports what it ran into, if anything
   * @return the thread which is doing the reconfiguration
   */
  private static Thread startReplayThreadReconfiguration(
      final Integer replayThreads, final AtomicReference<Throwable> failure)
  {
    final Thread reconfiguration = new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          setNumberOfReplayThreads(replayThreads);
        }
        catch (Throwable t)
        {
          failure.set(t);
        }
      }
    }, "replay thread reconfiguration");
    // A reconfiguration which never returns holds the monitor of MultimasterReplication:
    // let the fork end on it rather than have it kept alive by a thread of this test.
    reconfiguration.setDaemon(true);
    reconfiguration.start();
    return reconfiguration;
  }

  /**
   * Reads the acknowledgement of the provided change off the broker it was published on.
   * <p>
   * The messages which come first are discarded: this broker is told about everything the
   * replication server has for it, and the ack of one change is what is being looked for.
   *
   * @param broker the broker the change was published on
   * @param csn the change the ack is expected for
   * @return the ack of that change
   * @throws Exception if it never arrived
   */
  private static AckMsg awaitAck(final ReplicationBroker broker, final CSN csn) throws Exception
  {
    final long deadline = System.nanoTime() + SECONDS.toNanos(60);
    while (deadline - System.nanoTime() > 0)
    {
      final ReplicationMsg msg;
      try
      {
        msg = broker.receive();
      }
      catch (SocketTimeoutException e)
      {
        // The broker reads under a timeout of its own, which is far shorter than the
        // budget here: a quiet second is not an answer.
        continue;
      }
      if (msg == null)
      {
        // The broker stopped rather than timed out: there is nothing left to read from,
        // and reading it again would spin a core for the rest of the budget.
        throw new AssertionError("the session " + csn + " was published on is gone,"
            + " so the ack of that change can no longer arrive");
      }
      if (msg instanceof AckMsg && csn.equals(((AckMsg) msg).getCSN()))
      {
        return (AckMsg) msg;
      }
    }
    throw new AssertionError("the delivery of " + csn + " was never acknowledged");
  }

  /**
   * Waits for the delivery which took over from the abandoned one to be parked, reporting
   * what the reconfiguration ran into when that is why nothing was parked.
   * <p>
   * A reconfiguration which throws once it has stopped the replay threads leaves this
   * server with no replay thread at all: nothing can be parked then, and the timeout of
   * the wait would be reported in place of the failure which brought it about.
   *
   * @param parked the park the delivery is expected to be caught in
   * @param failure where the reconfiguration reports what it ran into, if anything
   * @return the thread which is replaying the parked operation
   * @throws Exception if no operation was parked in time
   */
  private static Thread awaitParkedOrReportReconfigurationFailure(
      final ParkedReplay parked, final AtomicReference<Throwable> failure) throws Exception
  {
    try
    {
      return parked.awaitParked(60, SECONDS);
    }
    catch (TimeoutException e)
    {
      final Throwable cause = failure.get();
      if (cause == null)
      {
        throw e;
      }
      final AssertionError error = new AssertionError(
          "the replay threads could not be reconfigured, so nothing was left to replay"
              + " the change which was handed back", cause);
      error.addSuppressed(e);
      throw error;
    }
  }

  /**
   * Changes the number of replay threads of this server, the way a change of
   * {@code num-update-replay-threads} does on a live server.
   *
   * @param replayThreads how many replay threads the server must run, or {@code null} for
   *          as many as it computes on its own
   */
  private static void setNumberOfReplayThreads(final Integer replayThreads)
  {
    final ReplicationSynchronizationProviderCfg cfg =
        mock(ReplicationSynchronizationProviderCfg.class);
    /*
     * An unstubbed mock hands out 0 replay threads and no connection timeout, and both
     * would outlive this test: the whole server shares one pool of replay threads. The
     * number of them is what this test is changing; the connection timeout is read back
     * from the running server, so that it is restored rather than restated.
     */
    when(cfg.getNumUpdateReplayThreads()).thenReturn(replayThreads);
    when(cfg.getConnectionTimeout())
        .thenReturn((long) MultimasterReplication.getConnectionTimeoutMS());
    multimasterReplication().applyConfigurationChange(cfg);
  }

  /** Returns the replication synchronization provider of the running server. */
  private static MultimasterReplication multimasterReplication()
  {
    // Read as an Object: the provider is declared with the configuration of its own type,
    // which is not the one the registry lists.
    for (Object provider : DirectoryServer.getSynchronizationProviders())
    {
      if (provider instanceof MultimasterReplication)
      {
        return (MultimasterReplication) provider;
      }
    }
    throw new AssertionError("this server runs no replication synchronization provider");
  }

  /**
   * Waits for the provided thread to be inside the {@code join()} of the replay threads it
   * is stopping.
   * <p>
   * Waiting for that, rather than for the thread to be started, is what puts the change in
   * the hands of a thread which has already been asked to stop: every replay thread is
   * asked to stop before the first of them is joined, so a thread which is joining has
   * asked the parked one. Where the thread waits is checked as well as that it waits: the
   * state on its own would be satisfied by any wait at all, including one taken before the
   * replay threads were asked to stop.
   *
   * @param reconfiguration the thread which is changing the number of replay threads
   * @param failure where that thread reports what it ran into, if anything
   * @throws Exception if it never reached the join
   */
  private static void awaitStoppingTheReplayThreads(
      final Thread reconfiguration, final AtomicReference<Throwable> failure) throws Exception
  {
    final long deadline = System.nanoTime() + SECONDS.toNanos(60);
    while (deadline - System.nanoTime() > 0)
    {
      if (reconfiguration.getState() == Thread.State.WAITING
          && isJoiningTheReplayThreads(reconfiguration))
      {
        return;
      }
      if (!reconfiguration.isAlive())
      {
        if (failure.get() != null)
        {
          throw new AssertionError(
              "the replay threads could not be reconfigured", failure.get());
        }
        fail("the replay threads were stopped without joining the one which holds a change");
      }
      // Sampling the stack of a running thread costs a handshake with it, so it is done
      // often enough to open the gate promptly and not so often as to slow the server
      // this test is watching: the thread it waits for is not going anywhere.
      Thread.sleep(10);
    }
    fail("the reconfiguration never reached the join() of the replay threads");
  }

  /** Returns whether the provided thread is waiting on the replay threads it stopped. */
  private static boolean isJoiningTheReplayThreads(final Thread reconfiguration)
  {
    for (final StackTraceElement frame : reconfiguration.getStackTrace())
    {
      if ("stopReplayThreads".equals(frame.getMethodName())
          && MultimasterReplication.class.getName().equals(frame.getClassName()))
      {
        return true;
      }
    }
    return false;
  }

  /**
   * Test case for [Issue 889]: every change which can not be replayed must be given up
   * on, not only the one which fails on its own.
   * <p>