[#922] Give a change back when the replay which owns it is unwound (#958)

Valery Kharseko

11 hours ago 776339a8c63c0bd8c83cfe8a619f6146794358e8

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

@@ -338,6 +338,17 @@
   * in flight unreplayable, and one alert per change would be a storm.
   */
  private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000;
  /**
   * What the ack of a delivery whose replay ran out of memory says did not apply the
   * change.
   * <p>
   * A constant rather than a message built from the change, because this is read on the
   * road out of an {@code OutOfMemoryError}: formatting one asks the JVM for the memory
   * it has just refused, and what {@code processUpdateDone()} needs of this string is
   * that there is one - it sets {@code hasReplayError} on the ack and never sends the
   * text, which is why no ordinal is spent on it either.
   */
  private static final String REPLAY_RAN_OUT_OF_MEMORY = "the replay of this change ran out of memory";
  /** The number of updates this replica gave up replaying. */
  private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger();
  /** Set while a replay thread is restarting the session after a failed replay. */
@@ -2564,6 +2575,11 @@

  /**
   * Create and replay a synchronized Operation from an UpdateMsg.
   * <p>
   * The change is given back on the way out of a replay which was unwound before it
   * reached one of the roads which give it back: a change which stays owned by a thread
   * which is not replaying it anymore is refused as a duplicate on every later delivery,
   * so this domain's ServerState would never move past it (issue #922).
   *
   * @param msg
   *          The UpdateMsg to be replayed.
@@ -2572,6 +2588,153 @@
   */
  void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
  {
    try
    {
      replayChangeAndTheChangesWaitingForIt(msg, replayThreadShutdown);
    }
    catch (Throwable t)
    {
      /*
       * The roads which run to their end give the change back themselves, so what is left
       * to give back here is a change whose replay was unwound over them: by an
       * OutOfMemoryError, which is left to end this thread, or by a throw from what the
       * replay runs once the ack of the delivery has been published - the give-back of the
       * change and the hand-out of the changes which were waiting for it are on that road.
       *
       * It takes the road of a failed replay rather than being handed back on the spot, so
       * that a change which keeps unwinding the replays it is given to is counted as
       * failing and is eventually given up on: handing it back bare would have this domain
       * ask for it, and restart its session for it, for as long as the server is up.
       *
       * The changes this thread parked as waiting for another change are left alone: they
       * are handed to whichever thread clears the change they are waiting for, and that
       * thread takes them over.
       *
       * Which change this thread owns is read before anything is done with it, and that
       * read takes no lock and allocates nothing: everything below is gated on the answer,
       * so a lookup which threw in its turn - on the road out of a JVM which has just
       * refused an allocation - would leave the change listed, uncommitted and owned by a
       * thread which is about to end, which is the state this whole issue is about.
       */
      CSN owned = null;
      try
      {
        owned = remotePendingChanges.getChangeOwnedByCurrentThread();
        if (owned != null)
        {
          if (replayThreadShutdown.get() || shutdown.get() || disabled)
          {
            /*
             * The replay was not failing, it was being abandoned: this thread is stopping
             * because their number is being changed, or the domain is going away or is
             * being imported into. The change is handed back without being counted against
             * its give-up budget, which is the road a replay abandoned that way takes.
             */
            abandonReplay(owned);
          }
          else
          {
            /*
             * A JVM which has run out of memory is told apart here rather than reported on
             * its own road: the change is given back counted, the way every other unwound
             * replay gives it back, but the line which says it is being asked for again is
             * not built - that asks the JVM for the memory it has just refused - and the
             * session is restarted without sitting through the backoff, since the thread
             * which is doing it is on its way out.
             *
             * A change whose budget is spent is still reported and still raises its alert
             * on this road: it is the one line which says this replica has diverged, and an
             * operator who is not told would be left with a replica which is silently
             * behind. That is the deliberate exception to the rule above.
             */
            recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError);
          }
        }
      }
      catch (Throwable recoveryFailure)
      {
        /*
         * The give-back is the road which hands the change over, so a throw out of it - an
         * allocation which fails in its turn, where what unwound the replay was a JVM out
         * of memory - would leave the change owned by this thread after all. Hand it back
         * bare, without the failure count that road did not reach, and restart the session
         * so that it is delivered again: this is the last resort, the throwable is rethrown
         * whatever happens here, and the thread this runs on may well be ending on it - so
         * a request left for the next recovery of this domain to pick up is a request which
         * may never be run.
         */
        if (owned != null)
        {
          remotePendingChanges.replayFailed(owned);
          sessionRestartRequested.set(true);
          /*
           * Reported and restarted under guards of their own, and in that order: the report
           * is the line an operator acts on, and the restart is what has the change
           * delivered again - so a report which can not be formatted, on a road an
           * OutOfMemoryError leads to, must not cost the restart.
           */
          try
          {
            logger.error(ERR_REPLAY_GIVE_BACK_FAILED, owned, getBaseDN(),
                stackTraceToSingleLineString(recoveryFailure));
          }
          catch (Throwable reportFailure)
          {
            suppress(recoveryFailure, reportFailure);
          }
          try
          {
            runRequestedSessionRestarts(false);
          }
          catch (Throwable restartFailure)
          {
            /*
             * Nothing is left to try: the change is listed, uncommitted and unowned, so any
             * later session restart of this domain delivers it again. This goes with the
             * throwable which is rethrown below rather than being reported on its own.
             */
            suppress(recoveryFailure, restartFailure);
          }
        }
        // The error which unwound the replay is the one reported, whatever the give-back
        // ran into on top of it.
        suppress(t, recoveryFailure);
      }
      throw t;
    }
  }

  /**
   * Records the second throwable as one the first suppressed, unless the two are one and
   * the same.
   * <p>
   * A JVM which has run out of memory hands out the error it prepared before it ran out as
   * often as it is asked for one, so the roads out of a replay can carry the same instance
   * twice - and a throwable can not suppress itself.
   *
   * @param thrown the throwable which is reported
   * @param alsoThrown what was met on the way out of it
   */
  private static void suppress(Throwable thrown, Throwable alsoThrown)
  {
    if (thrown != alsoThrown)
    {
      thrown.addSuppressed(alsoThrown);
    }
  }

  /**
   * Replays the change of the provided message, then the changes which were waiting for
   * it, for as long as there are some.
   *
   * @param msg
   *          The UpdateMsg to be replayed.
   * @param replayThreadShutdown
   *          whether the replay thread was asked to stop
   */
  private void replayChangeAndTheChangesWaitingForIt(
      LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
  {
    // Try replay the operation, then flush (replaying) any pending operation
    // whose dependency has been replayed until no more left.
    do
@@ -2582,6 +2745,13 @@
      boolean replayAbandoned = false;
      String replayErrorMsg = null;
      CSN csn = null;
      /*
       * Read once, before anything of this delivery has run, so that the report of an ack
       * which could not be published names the change even when reading it off the message
       * is what threw: that report is written on the way out of a road which is already
       * failing, and it must not be the throw which unwinds the replay.
       */
      final CSN delivered = msg.getCSN();
      try
      {
        // The next operation for which to attempt replay.
@@ -2922,7 +3092,57 @@
          replayErrorMsg = message.toString();
          replayFailed = true;
        }
      } finally
      }
      catch (OutOfMemoryError e)
      {
        /*
         * The JVM is out of memory, which is not something to carry on replaying from: the
         * error is left to unwind the replay thread, which ends on it (issue #923). It is
         * not turned into a report of its own here either - the stack trace of an error
         * which can be raised anywhere says little, and the uncaught exception handler of
         * DirectoryThread writes the one line this is worth, with an alert.
         *
         * The other errors of the JVM take the road below. A StackOverflowError is met by
         * the thread which recursed and is gone once the stack has unwound, and the entry
         * being replayed is what raises it rather than the state of this server: ending a
         * thread on it would have one change this replica can not replay cost it a replay
         * thread per delivery, and nothing creates a replay thread to replace one which
         * ends.
         *
         * The change is given back, counted as failing and asked for again on the way out
         * (issue #922). The one thing done here is to make the ack this delivery publishes
         * below say that the change was not applied: a replica which is asking for a change
         * again must not have told an assured write that it is in the data here. It is a
         * constant rather than a message built from the change, because building one asks
         * the JVM for the memory it has just refused.
         */
        replayErrorMsg = REPLAY_RAN_OUT_OF_MEMORY;
        throw e;
      }
      catch (Error e)
      {
        /*
         * An Error out of the replay - a LinkageError met where a plugin or a backend class
         * is loaded, an AssertionError - unwinds every road this replay has out of here, the
         * ones which give the change back among them. The change is not in the data, so it
         * is reported and given back here, on the road every other failed replay takes: the
         * ack below says it was not applied, the failure counts against the give-up budget
         * of the change, the session is restarted for it to be delivered again (issue #922)
         * and the changes which were waiting for this one are replayed rather than left
         * waiting for a thread to hand them out.
         *
         * It is not given up on where no operation could be built from the message, which is
         * what an Exception at that point means: an Error says that this server could not run
         * the replay, not that the message is one no delivery could ever build an operation
         * from.
         */
        final LocalizableMessage message = ERR_ERROR_REPLAYING_CHANGE.get(
            msg.getCSN(), getBaseDN(), stackTraceToSingleLineString(e));
        logger.error(message);
        replayErrorMsg = message.toString();
        replayFailed = true;
      }
      finally
      {
        if (!dependency)
        {
@@ -2934,7 +3154,88 @@
           * down, so nothing would reach the server which is waiting for it, and an
           * assured write would wait out its timeout rather than be told what happened.
           */
          processUpdateDone(msg, replayErrorMsg);
          try
          {
            processUpdateDone(msg, replayErrorMsg);
          }
          catch (OutOfMemoryError e)
          {
            /*
             * The one throw from here which is not caught, on the same terms as the arm
             * above: a JVM which has run out of memory is not something to carry on
             * replaying from, the error is left to end this replay thread, and the uncaught
             * exception handler of DirectoryThread raises the alert #923 is about. Swallowed
             * here instead, it would have this thread go on to the roads below and to the
             * next change of the loop on an exhausted heap, with nothing reported anywhere.
             *
             * What the give-back on the way out of replay() then finds depends on the road
             * the replay took to get here. A replay which failed, or which was unwinding on
             * an OutOfMemoryError of its own, still owns its change: it is given back
             * counted, and the thread ends on this error rather than on the one it stepped
             * over. A replay which committed owns nothing anymore - commit() cleared the
             * owner, and the index the give-back reads, in the same step - so the give-back
             * is a no-op, and rightly so: a change which is in the data is not one to ask
             * for again. What that road steps over is getNextUpdate() below, so the changes
             * parked behind the committed change wait for the next replay of this domain to
             * hand them out. That is the trade #923 asks for: a thread which met this error
             * is not to carry on, not even for them.
             */
            throw e;
          }
          catch (Throwable ackFailure)
          {
            /*
             * Publishing the ack says nothing about whether the change was applied, so a
             * throw here must not be a road out of the replay. It would step over the
             * give-back of the change and, where the change was committed, over the
             * getNextUpdate() below - the one drain of the changes this thread parked as
             * waiting for it - leaving them waiting for a thread which is not replaying
             * anything anymore.
             *
             * A master which is waiting for the ack waits out its assured timeout either
             * way - and there is one only for an assured write in safe-read mode, which is
             * the one delivery a replica acknowledges. processUpdateDone() runs for every
             * delivery all the same: it accounts for the delivery in the receive window and
             * in the processed-updates counter, and a throw from there is a throw from this
             * bookkeeping, with no ack owed to anybody. Either way it is reported and the
             * replay carries on to the road the change itself decided: applied, failed and
             * asked for again, or given up on.
             *
             * Every step of processUpdateDone() catches what it can meet - the broker keeps
             * a failure to publish to itself and retries it - so what reaches here is what
             * no code of the replication protocol expected: an Error, and the tests drive
             * it as one.
             */
            try
            {
              logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(),
                  stackTraceToSingleLineString(ackFailure));
            }
            catch (Throwable reportFailure)
            {
              /*
               * Guarded like the report of a give-back which failed: this one runs on the
               * same kind of road - the stack of the throwable is walked to build the line -
               * and a report which can not be built must not become the throw which unwinds
               * the replay past the give-back and past getNextUpdate().
               *
               * The line is tried once more with the name of the error alone, which walks no
               * stack. Nothing rethrows what was caught here, so an error recorded as
               * suppressed on it would be recorded nowhere, and this is the road on which an
               * operator has the least to go on. A second refusal leaves nothing to say it
               * with, and the replay carries on with what the change decided.
               */
              try
              {
                logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(),
                    ackFailure.getClass().getName());
              }
              catch (Throwable secondReportFailure)
              {
                // Nothing is left to say it with.
              }
            }
          }
        }
      }

@@ -3191,6 +3492,29 @@
   */
  private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean replayThreadShutdown)
  {
    return recoverFromReplayFailure(csn, replayThreadShutdown, false);
  }

  /**
   * Recovers from a change which could not be replayed, telling apart the replay which was
   * unwound by a JVM out of memory.
   *
   * @param csn
   *          the CSN of the change which could not be replayed
   * @param replayThreadShutdown
   *          whether the replay thread was asked to stop
   * @param outOfMemory
   *          whether what unwound the replay was the JVM running out of memory, in which
   *          case the line which says the change is being asked for again is not built -
   *          the report of a change which is given up on is, since it is what says this
   *          replica has diverged - and the session is restarted without the backoff
   * @return {@code true} when the caller must stop replaying because the session is
   *         being restarted or is going away, {@code false} when it may carry on with
   *         the changes which follow
   */
  private boolean recoverFromReplayFailure(
      CSN csn, AtomicBoolean replayThreadShutdown, boolean outOfMemory)
  {
    /*
     * The failure is recorded, and the change given up on, while this thread still owns
     * it: a change which is listed, uncommitted and unowned is what putRemoteUpdate()
@@ -3254,7 +3578,16 @@
      return true;
    }

    logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
    if (!outOfMemory)
    {
      /*
       * Not on the road out of a JVM which has run out of memory: building this line asks
       * it for the memory it has just refused, and the ack of the delivery already says
       * that the change was not applied. The constant that ack carries exists for the same
       * reason.
       */
      logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
    }
    /*
     * This change is not owned by anyone anymore, so the session has to be restarted for
     * the replication server to deliver it again. Ask for the restart before trying to
@@ -3267,9 +3600,12 @@
     * A replay thread which is stopping - the number of them is being changed - restarts
     * the session all the same: nothing else would ask for the change it just released,
     * and the ServerState would stay behind it for good. It does not sit through the
     * backoff on its way out, though: the backend is not what is going away.
     * backoff on its way out, though: the backend is not what is going away. Neither does
     * the thread an OutOfMemoryError is ending, for the same reason - and the restart is
     * run rather than left to be asked for again, because that thread will not be there to
     * run it, and a change nobody asks for again holds this domain's ServerState back.
     */
    runRequestedSessionRestarts(!replayThreadShutdown.get());
    runRequestedSessionRestarts(!replayThreadShutdown.get() && !outOfMemory);
    return true;
  }

@@ -3292,7 +3628,41 @@
      {
        while (sessionRestartRequested.getAndSet(false))
        {
          restartSession(wait);
          boolean restarted = false;
          try
          {
            restartSession(wait);
            restarted = true;
          }
          finally
          {
            if (!restarted)
            {
              /*
               * The request is put back where it was taken from. The flag is read and
               * cleared before the restart runs, so a restart which ends abruptly - the
               * session is stopped first, and starting it again creates a listener thread,
               * which the operating system can refuse - would otherwise leave this domain
               * with no session and with nothing left to ask for one.
               *
               * What a request left standing buys is bounded, and the bound is worth
               * stating. Its two readers are the roads out of a failed and of an abandoned
               * replay of this domain, and with no listener thread nothing is delivered
               * anymore: the replays left to run are the changes already taken off the
               * session - the ones waiting in the replay queue, and the ones parked as
               * dependencies. One of those failing finds the request standing and runs the
               * restart, which starts from a clean state, since disableService() drops the
               * listener thread which was never started. Once they are spent, the domain
               * stays down until it is disabled and enabled back, or the server is
               * restarted. That is said where it can be heard: a refused thread is an
               * OutOfMemoryError, and one which leaves recoverFromReplayFailure() or
               * abandonReplay() ends the replay thread it is met on, so the uncaught
               * exception handler of DirectoryThread writes the line and raises the alert,
               * with the start of the listener thread in the trace.
               */
              sessionRestartRequested.set(true);
            }
          }
        }
      }
      finally

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

@@ -17,6 +17,8 @@
 */
package org.opends.server.replication.plugin;

import net.jcip.annotations.GuardedBy;

import org.opends.server.replication.common.CSN;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -36,11 +38,18 @@
   */
  private volatile UpdateMsg msg;
  /**
   * Whether a replay thread owns this change: it is being replayed, or it waits for the
   * The replay thread which owns this change: it is being replayed, or it waits for the
   * change it depends on. A remote change which no thread owns is one whose replay
   * failed and which the replication server is expected to deliver again.
   * <p>
   * The owner is kept rather than the bare fact that there is one, so that a change is
   * given back by the thread it was handed to and by nobody else: a release which arrives
   * from a thread which does not own the change anymore - it reports a failure on a change
   * which has been taken over since - would hand a change which is being replayed right
   * now to a second thread (issue #922).
   */
  private boolean owned;
  @GuardedBy("RemotePendingChanges.pendingChangesLock")
  private Thread owner;
  /**
   * How many times in a row the replay of this change failed, and when the first of
   * those failures happened - on a clock which only moves forward.
@@ -138,18 +147,30 @@
   */
  public boolean isOwned()
  {
    return owned;
    return owner != null;
  }

  /**
   * Sets whether a replay thread owns this change.
   * Returns whether the provided thread owns this change.
   *
   * @param owned {@code true} when a replay thread takes the change over, {@code false}
   *              when its replay failed and the change must be delivered again
   * @param thread the thread which claims the change
   * @return {@code true} if that thread is the one this change was handed to
   */
  public void setOwned(boolean owned)
  public boolean isOwnedBy(Thread thread)
  {
    this.owned = owned;
    // A change nobody owns is not owned by a caller which has no thread to name either.
    return thread != null && owner == thread;
  }

  /**
   * Sets the replay thread which owns this change.
   *
   * @param owner the thread which takes the change over, or {@code null} when it is given
   *              back - its replay failed, or it has been applied
   */
  public void setOwner(Thread owner)
  {
    this.owner = owner;
  }

  /**

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

@@ -23,6 +23,8 @@
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -73,6 +75,32 @@
   */
  private final ConcurrentSkipListSet<PendingChange> activeAndDependentChanges = new ConcurrentSkipListSet<>();

  /**
   * The change each replay thread is replaying right now, read by the give-back on the way
   * out of a replay which was unwound.
   * <p>
   * It is an index of what {@link PendingChange#isOwnedBy(Thread)} already says rather than
   * a second copy of it: it is written in the same locked step wherever a change is taken
   * over or given back, and every road which acts on what it answers checks the ownership
   * of the change again. What it buys is the read. That read is made on a road an
   * {@code OutOfMemoryError} leads to, and looking the change up by walking the pending
   * changes takes two locks and allocates an iterator - an allocation a JVM which has just
   * refused one may well refuse again, and the change would then be left listed,
   * uncommitted and owned by a thread which is not replaying it anymore, which is the wedge
   * this issue is about (issue #922).
   * <p>
   * A thread is entered here when it takes a change over and removed when it gives it back,
   * applies it, or parks it as waiting for another change - the parked ones are handed to
   * whichever thread clears what they wait for, so they are not this one's to give back.
   * <p>
   * The entry of a thread is written by that thread and by nobody else, and that - not the
   * lock - is what keeps the writes apart: the park in {@link #addDependency(PendingChange)}
   * clears it under the read lock, where every other writer holds the write lock, and it
   * races nothing for it. {@link #clear()} is the one writer of every entry, and it holds
   * both locks.
   */
  private final ConcurrentMap<Thread, CSN> changeBeingReplayed = new ConcurrentHashMap<>();

  private final ReentrantReadWriteLock pendingChangesLock = new ReentrantReadWriteLock(true);
  private final ReentrantReadWriteLock.ReadLock pendingChangesReadLock = pendingChangesLock.readLock();
  private final ReentrantReadWriteLock.WriteLock pendingChangesWriteLock = pendingChangesLock.writeLock();
@@ -209,9 +237,25 @@

  /**
   * Mark an update message as committed.
   * <p>
   * A change another replay thread owns is not this thread's to record: it is reported as
   * a change which is not here, the way one which is not listed anymore is. A thread
   * decides that a change is in the data, or that it is to be given up on, and records
   * that decision a turn of this lock later - long enough for the change to have been
   * handed back, delivered again and taken over in between. Recording it then would
   * advance the ServerState over a change which is not in the data yet (issue #889) and
   * have the thread which is applying it right now fail to commit.
   * <p>
   * The give-back on the way out of an unwound replay is what makes this reachable: it
   * runs wherever the replay was left, so the checks which keep it from taking a change
   * away from the thread which owns it now belong on every road which reads ownership,
   * not only on {@link #replayFailed(CSN)} (issue #922).
   *
   * @param csn
   *          The CSN of the update message that must be set as committed.
   * @throws NoSuchElementException
   *          if there is no change with that CSN for this thread to record: it is not
   *          listed as pending anymore, or another replay thread owns it
   */
  public void commit(CSN csn)
  {
@@ -219,12 +263,14 @@
    try
    {
      PendingChange curChange = pendingChanges.get(csn);
      if (curChange == null)
      if (curChange == null
          || (curChange.isOwned() && !curChange.isOwnedBy(Thread.currentThread())))
      {
        throw new NoSuchElementException();
      }
      curChange.setCommitted(true);
      curChange.setOwned(false);
      curChange.setOwner(null);
      changeBeingReplayed.remove(Thread.currentThread(), csn);
      activeAndDependentChanges.remove(curChange);

      final Iterator<PendingChange> it = pendingChanges.values().iterator();
@@ -271,6 +317,12 @@
   * The changes another replay thread is applying right now are left alone: they are
   * about to commit, and forgetting them would have their {@code commit()} fail, the
   * ServerState stay behind them and the replication server replay them a second time.
   * <p>
   * Only the thread which owns the change gives it back. A release which arrives from
   * another one is a release of a change which has been taken over since - the failure of
   * a delivery is reported after the change it carried was handed to the delivery which
   * follows it - and taking the change away from the thread which is replaying it right
   * now is the double replay the ownership is there to prevent (issue #922).
   *
   * @param csn the CSN of the change whose replay failed
   */
@@ -280,9 +332,10 @@
    try
    {
      final PendingChange change = pendingChanges.get(csn);
      if (change != null && !change.isCommitted())
      if (change != null && !change.isCommitted() && change.isOwnedBy(Thread.currentThread()))
      {
        change.setOwned(false);
        change.setOwner(null);
        changeBeingReplayed.remove(Thread.currentThread(), csn);
      }
    }
    finally
@@ -339,9 +392,13 @@
   *          the CSN of the change whose replay failed
   * @param nowMs
   *          when it failed, on a clock which only moves forward
   * @return the failures of the change, or {@code null} when it is not listed as an
   *         uncommitted change anymore, which happens when the domain was disabled while
   *         it was being replayed: there is no change left here to give up on
   * @return the failures of the change, or {@code null} when there is no change left here
   *         for this thread to give up on: it is not listed as an uncommitted change
   *         anymore, which happens when the domain was disabled while it was being
   *         replayed, or another replay thread owns it, which is that change having been
   *         delivered again and taken over while this thread was on its way to reporting
   *         on it. Spending the give-up budget of a change another thread is applying
   *         would have this replica skip a change which is being written (issue #922)
   */
  public ReplayFailure recordReplayFailure(CSN csn, long nowMs)
  {
@@ -349,7 +406,8 @@
    try
    {
      final PendingChange change = pendingChanges.get(csn);
      if (change == null || change.isCommitted())
      if (change == null || change.isCommitted()
          || (change.isOwned() && !change.isOwnedBy(Thread.currentThread())))
      {
        return null;
      }
@@ -434,6 +492,7 @@
      pendingChanges.clear();
      dependentChanges.clear();
      activeAndDependentChanges.clear();
      changeBeingReplayed.clear();
      failingChanges = 0;
    }
    finally
@@ -465,8 +524,16 @@
      {
        return false;
      }
      change.setOwned(true);
      /*
       * Listed as being replayed before it is owned, and not the other way round: the
       * caller enters the replay - where the give-back on the way out lives - once this
       * returns, so nothing which allocates must run between the owner being stamped and
       * that. A change listed here without an owner is the state a failed replay leaves
       * behind, and the dependency checks which read this set do not read the owner.
       */
      activeAndDependentChanges.add(change);
      changeBeingReplayed.put(Thread.currentThread(), change.getCSN());
      change.setOwner(Thread.currentThread());
      return true;
    }
    finally
@@ -475,7 +542,40 @@
    }
  }
  /**
   * Returns the CSN of the change the calling thread is replaying, when it still owns one.
   * <p>
   * A thread owns the change it is replaying and the ones it parked as waiting for another
   * change. The parked ones are left out: they are handed to whichever thread clears the
   * change they are waiting for, and that thread takes them over, so giving one back here
   * would have the same change handed to two threads (issue #922).
   * <p>
   * It is a plain read of {@link #changeBeingReplayed}: no lock is taken and nothing is
   * allocated. This is what the give-back on the way out of an unwound replay asks first,
   * and it runs on a road an {@code OutOfMemoryError} leads to - a lookup which allocated
   * could be refused in its turn, and a give-back which does not know which change to give
   * back leaves it listed, uncommitted and owned by a thread which is not replaying it
   * anymore, which is the wedge this issue is about.
   * <p>
   * An answer which is out of date is safe: every road which acts on it - {@code commit()},
   * {@link #recordReplayFailure(CSN, long)} and {@link #replayFailed(CSN)} - checks the
   * ownership of the change again under the write lock, and is a no-op for a change this
   * thread does not own anymore.
   *
   * @return the CSN of the change this thread is replaying, or {@code null} when it does
   *         not own one anymore - the road its replay took gave it back, or it was applied
   */
  CSN getChangeOwnedByCurrentThread()
  {
    return changeBeingReplayed.get(Thread.currentThread());
  }

  /**
   * Get the first update in the list that have some dependencies cleared.
   * <p>
   * The change is handed to the calling thread, which owns it from then on: it is
   * replayed by whichever replay thread cleared the change it was waiting for rather than
   * by the one which parked it, and a change is given back by the thread which owns it
   * and by nobody else (issue #922).
   *
   * @return The LDAPUpdateMsg to be handled.
   */
@@ -485,22 +585,65 @@
    dependentChangesLock.lock();
    try
    {
      if (!dependentChanges.isEmpty() && !pendingChanges.isEmpty())
      if (!hasChangeToHandOut())
      {
        PendingChange firstDependentChange = dependentChanges.first();
        if (pendingChanges.firstKey().isNewerThanOrEqualTo(firstDependentChange.getCSN()))
        {
          dependentChanges.remove(firstDependentChange);
          return firstDependentChange.getLDAPUpdateMsg();
        }
        /*
         * Nothing is waiting, or what waits is still held back by the changes before it.
         * This is called at the end of every replay, by every replay thread, so the answer
         * is looked for under the read lock: taking the write lock here would have a
         * backlog of waiting changes serialize the replay of the changes which have none.
         */
        return null;
      }
      return null;
    }
    finally
    {
      dependentChangesLock.unlock();
      pendingChangesReadLock.unlock();
    }

    /*
     * There is one to hand out, and handing it out writes its owner, which is written
     * under the write lock as the rest of the state of a change is. It is looked for again
     * under that lock: another replay thread may have been handed it in between.
     */
    pendingChangesWriteLock.lock();
    dependentChangesLock.lock();
    try
    {
      if (hasChangeToHandOut())
      {
        final PendingChange firstDependentChange = dependentChanges.first();
        /*
         * Entered as the change this thread is replaying before it is taken out of the
         * ones which are waiting, for the same reason markInProgress() enters it before it
         * stamps the owner: an allocation which fails here must leave the change where it
         * was rather than take it out of the hands which would hand it out again.
         */
        changeBeingReplayed.put(Thread.currentThread(), firstDependentChange.getCSN());
        dependentChanges.remove(firstDependentChange);
        firstDependentChange.setOwner(Thread.currentThread());
        return firstDependentChange.getLDAPUpdateMsg();
      }
      return null;
    }
    finally
    {
      dependentChangesLock.unlock();
      pendingChangesWriteLock.unlock();
    }
  }

  /**
   * Returns whether the first change waiting for another one can be replayed now, that is
   * whether every change before it has left the pending changes.
   */
  @GuardedBy("pendingChangesLock, dependentChangesLock")
  private boolean hasChangeToHandOut()
  {
    return !dependentChanges.isEmpty()
        && !pendingChanges.isEmpty()
        && pendingChanges.firstKey().isNewerThanOrEqualTo(dependentChanges.first().getCSN());
  }

  /**
@@ -524,6 +667,15 @@
      {
        dependentChanges.add(dependentChange);
      }
      /*
       * Whichever of the two it was, this thread is not replaying that change anymore: a
       * parked one is handed to the thread which clears what it waits for, and one which is
       * not listed here anymore is gone with the pending changes of a domain which was
       * disabled. The owner stays as it is - it is what has getNextUpdate() hand the change
       * over rather than leave it to nobody - and the give-back on the way out of an
       * unwound replay leaves it alone (issue #922).
       */
      changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
    }
    finally
    {

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

@@ -116,13 +116,33 @@
          domain.replay(updateMsg, shutdown);
        }
      }
      catch (Exception e)
      catch (OutOfMemoryError e)
      {
        /*
         * The JVM is out of memory, which is not something to carry on replaying from: this
         * thread does not stay for the changes which follow. Nothing is reported here - the
         * uncaught exception handler of DirectoryThread is what says this thread is gone,
         * with an alert - and the change it was replaying has been given back, counted and
         * asked for again by the domain on its way out (issue #922).
         *
         * The other errors of the JVM are caught below: a StackOverflowError is gone once
         * the stack has unwound, and a thread which ends here is one nothing replaces.
         */
        throw e;
      }
      catch (Throwable t)
      {
        /*
         * catch all exceptions happening so that the thread never dies even
         * in case of problems.
         *
         * An Error is not an Exception, so one raised here used to unwind run() and end
         * this thread. Nothing creates a replay thread to replace it - the pool is created
         * when the first domain of this server is - so the shared replay queue would have
         * one consumer fewer for every domain, for as long as the server is up, until it
         * has none left and replication stops (issue #923).
         */
        logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(e));
        logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(t));
      }
    }
    if (logger.isTraceEnabled())

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -657,6 +657,17 @@
 to each other at the same time, of which one is dropped while the other one serves the domain. Until \
 a handshake completes, no change is replicated over this connection, and the connection which \
 completes one is reported in its turn
ERR_ERROR_REPLAYING_CHANGE_315=An Error was thrown while replaying change %s in domain "%s": %s. \
 The change has not been recorded as replayed and is given back to the replication server, which \
 still owns it and sends it again
ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain "%s" once its \
 replay was done - that is where the delivery is accounted for and, for an assured write in \
 safe-read mode, acknowledged: %s. Whether the change was applied is what its replay decided, \
 and it is unaffected. A server which is waiting for that acknowledgement waits out its assured \
 timeout instead
ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \
 server after the replay which owned it was unwound: %s. The change has been released without its \
 failure being counted, and the session is being restarted so that the change is delivered again
WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \
 the replay of one of its changes to finish. A change which reaches the backend from now on \
 is not recorded in the ServerState being saved, so the replication server sends it again \

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

@@ -31,6 +31,7 @@
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.function.Supplier;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -246,6 +247,9 @@
      park.deregister();
    }
    parks.clear();
    // Same shape: a throw which outlives the test which asked for it unwinds the replays of
    // every test which follows, for as long as this plugin is loaded.
    replayThrows.clear();
  }


@@ -661,6 +665,23 @@
     */
    if (operation.isSynchronizationOperation())
    {
      /*
       * An error thrown here is thrown from inside the run() of the operation, which is
       * where a plugin or a backend class which can not be loaded raises one: it unwinds
       * the replay the way a real one does, rather than being reported as a result code the
       * replay decides on. It comes before the park because it is the whole point of the
       * delivery which asked for it, and no test asks for both on one operation.
       */
      final ThrownFromReplay thrower = replayThrows.get(key);
      if (thrower != null)
      {
        final Error error = thrower.errorFor(operation);
        if (error != null)
        {
          throw error;
        }
      }

      final ParkedReplay park = parks.get(key);
      if (park != null && park.parks(operation))
      {
@@ -786,6 +807,82 @@
  /** Registered parks for the replayed operations, keyed like the short circuits. */
  private static final Map<String, ParkedReplay> parks = new ConcurrentHashMap<>();

  /** The errors the replayed operations of one type throw, by operation type and section. */
  private static final Map<String, ThrownFromReplay> replayThrows = new ConcurrentHashMap<>();

  /**
   * Throws an error out of the replay of the operations of one type, as many times as the
   * test asked for.
   * <p>
   * The throw is made at a plugin point which runs inside {@code op.run()}, so it unwinds
   * the replay from where a plugin or a backend class which can not be loaded raises one -
   * past the point where the change was marked as being replayed by the thread which took
   * it. That is what tells it apart from a delivery which reports a result code: a result
   * code is a verdict the replay decided on, an error is the replay not running at all.
   * <p>
   * It is bounded rather than standing: the delivery which takes over from the one which
   * was unwound has to be able to apply the change, or the test would watch this replica
   * give up on a change it was never going to replay.
   */
  public static final class ThrownFromReplay
  {
    private final String key;
    /** Which of the replayed operations of that type this throws out of. */
    private final Predicate<PluginOperation> matches;
    /** Built where it is thrown, so that it carries the stack of the replay it unwound. */
    private final Supplier<? extends Error> error;
    private final int maxTimes;
    private final AtomicInteger thrown = new AtomicInteger();

    private ThrownFromReplay(String key, Predicate<PluginOperation> matches,
        Supplier<? extends Error> error, int maxTimes)
    {
      this.key = key;
      this.matches = matches;
      this.error = error;
      this.maxTimes = maxTimes;
    }

    /**
     * Returns the error to throw out of the provided operation, or {@code null} when this
     * is not one of the operations it is for, or when it has been thrown as many times as
     * the test asked for.
     */
    private Error errorFor(PluginOperation operation)
    {
      if (!matches.test(operation))
      {
        return null;
      }
      // Claimed before it is built, so that two replays of one change - the retry in place
      // makes them - never take the same one twice.
      if (thrown.incrementAndGet() > maxTimes)
      {
        return null;
      }
      return error.get();
    }

    /**
     * Returns how many replays this threw out of.
     *
     * @return the number of replays which were unwound by this
     */
    public int thrownCount()
    {
      return Math.min(thrown.get(), maxTimes);
    }

    /**
     * Stops throwing out of the replayed operations. A test must call this however it ends,
     * or it leaves the deliveries of the tests which follow being unwound.
     */
    public void deregister()
    {
      replayThrows.remove(key, this);
    }
  }

  /**
   * Holds the replayed operations of one type where they are, one at a time, until the
   * test lets each of them go.
@@ -1076,6 +1173,40 @@
    return park;
  }

  /**
   * Throws the provided error out of the replay of the operations of the given type, at the
   * given plugin point, as many times as asked for and no more.
   *
   * @param operation the type of operation to throw out of
   * @param section the plugin point to throw at, which can only be {@code PreParse}
   * @param matches which of them to throw out of - the change a test acts on rather than
   *          whatever of that type reaches this point first
   * @param error builds the error where it is thrown, so that it carries the stack trace of
   *          the replay it unwound
   * @param maxTimes how many replays to unwind, after which the operations are let through:
   *          the delivery which takes over from the one which was unwound is what applies
   *          the change
   * @return the throw, which the test must {@link ThrownFromReplay#deregister()} when it is
   *         done with it
   * @throws IllegalArgumentException if asked for any plugin point but {@code PreParse}
   */
  public static ThrownFromReplay throwFromReplayedOperations(OperationType operation,
      String section, Predicate<PluginOperation> matches, Supplier<? extends Error> error,
      int maxTimes)
  {
    if (!"PreParse".equalsIgnoreCase(section))
    {
      // The pre-operation plugins are not invoked for synchronization operations at all, so
      // a throw asked for anywhere else is one no replay would ever meet.
      throw new IllegalArgumentException("replayed operations can only be thrown out of at"
          + " PreParse, which is the only plugin point they reach, not at " + section);
    }
    final String key = keyFor(operation, section);
    final ThrownFromReplay thrower = new ThrownFromReplay(key, matches, error, maxTimes);
    replayThrows.put(key, thrower);
    return thrower;
  }

  /** Returns the key a short circuit or a park of the given operations is kept under. */
  private static String keyFor(OperationType operation, String section)
  {

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

@@ -33,9 +33,12 @@
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -58,6 +61,7 @@
import org.opends.server.plugins.PausePreParsePlugin;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.plugins.ShortCircuitPlugin.ParkedReplay;
import org.opends.server.plugins.ShortCircuitPlugin.ThrownFromReplay;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
@@ -2584,13 +2588,738 @@
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed"));

    assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseOperationRefusesAControl::new,
        18, "user.889.7", "the replay must fail once the operation is built");
  }

  /**
   * The errors a replay meets which say nothing about the changes which follow: a class
   * which could not be linked, and a stack which ran out on the entry being replayed. The
   * second is an error of the JVM, but the thread which met it is whole again once the
   * stack has unwound and the entry is what raised it, so ending the thread would have one
   * change this replica can not replay cost it one replay thread per delivery.
   * <p>
   * Each row builds its error where it is thrown rather than here, so that the stack trace
   * it carries is the one of the replay it unwound.
   */
  @DataProvider(name = "recoverableReplayErrors")
  public Object[][] recoverableReplayErrors()
  {
    return new Object[][] {
      { (Supplier<Error>) () -> new LinkageError("the replay of this change meets an Error"),
        19, "user.922.1", "a LinkageError" },
      { (Supplier<Error>) StackOverflowError::new, 23, "user.922.5", "a StackOverflowError" },
    };
  }

  /**
   * Test case for [Issue 922] and [Issue 923]: a change whose replay threw an Error is
   * given back, and the thread which was replaying it is still there to take the changes
   * which follow.
   * <p>
   * A change is owned by the replay thread which took it, and that ownership is what
   * keeps a change being replayed from being replayed a second time (OPENDJ-1115): every
   * later delivery of it is refused as a duplicate. Ownership is given back on the roads
   * which run to their end, so an Error - which unwinds the replay out of every one of
   * them - would leave the change listed, uncommitted and owned by a thread which is not
   * replaying it anymore: nobody could replay it, and this domain's ServerState would
   * never move past it again.
   * <p>
   * The error is thrown where the operation is built, so what these rows pin is the arm
   * which reports it and takes the road of a failed replay: it is caught inside the replay
   * and never leaves it. The roads which do leave it - the give-back of a replay which was
   * unwound, and the widened catch of the replay thread which is what keeps that thread
   * alive - are pinned by
   * {@link #aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain()}, where the throw is
   * made past the point any catch of the replay runs on.
   */
  @Test(dataProvider = "recoverableReplayErrors")
  public void aChangeWhoseReplayThrewAnErrorIsDeliveredAgain(
      final Supplier<Error> error, int serverId, String uid, String what) throws Exception
  {
    testSetUp("aChangeWhoseReplayThrewAnErrorIsDeliveredAgain." + uid);
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseReplayThrewAnErrorIsDeliveredAgain " + uid));

    assertChangeIsDeliveredAgainAfter(
        (csn, dn, mods, entryUUID) ->
            new ModifyMsgWhoseReplayThrows(csn, dn, mods, entryUUID, error),
        serverId, uid, "the replay of this change throws " + what);
  }

  /**
   * Test case for [Issue 922] and [Issue 923]: the change a replay thread was replaying
   * when the JVM ran out of memory is given back before the error is left to end that
   * thread.
   * <p>
   * An OutOfMemoryError is not turned into a failed replay and reported the way the other
   * errors are - building the report asks for more of what the JVM has run out of - and
   * the thread it unwinds is not replaced. The change it was replaying must not go with
   * it: it is handed back so that the delivery which follows can replay it.
   */
  @Test
  public void aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain"));

    /*
     * The threads are read by identity rather than counted: what this test is about is the
     * thread which met the error being gone, which is what #923 sanctions. Whether the pool
     * is refilled afterwards is the half of that issue which is left open, and a count
     * would freeze it here as the behaviour which is wanted.
     */
    final Set<Long> replayThreadsBefore = replayThreadIds();
    final int initialUncaughtAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION);
    assertChangeIsDeliveredAgainAfter(
        (csn, dn, mods, entryUUID) -> new ModifyMsgWhoseReplayThrows(csn, dn, mods, entryUUID,
            () -> new OutOfMemoryError("the JVM is out of memory")),
        20, "user.922.2", "the replay of this change meets an OutOfMemoryError");
    /*
     * The change is given back, and so replayed by another thread, while the thread which
     * met the error is still unwinding - through the session restart, and then through the
     * uncaught exception handler which raises the alert - so its end is waited for rather
     * than read off the pool the moment the change lands. The alert is raised by that
     * handler once run() has returned, so it lands after the thread is gone, and it is
     * waited for in the same breath.
     */
    TestTimer timer = new TestTimer.Builder()
      .maxSleep(30, SECONDS)
      .sleepTimes(200, MILLISECONDS)
      .toTimer();
    timer.repeatUntilSuccess(new CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        assertFalse(replayThreadIds().containsAll(replayThreadsBefore),
            "an OutOfMemoryError must end the replay thread which met it");
        assertUncaughtExceptionAlertRaisedSince(initialUncaughtAlerts,
            "the replay thread an OutOfMemoryError ended must have raised the alert #923 asks"
                + " for on its way out");
      }
    });
  }

  /**
   * Asserts that a {@code DirectoryThread} has ended on an uncaught throwable since the
   * provided count of those alerts was read: the uncaught exception handler of its thread
   * group is what raises it, and it is the one line a replay thread which an
   * OutOfMemoryError ended leaves behind (issue #923).
   * <p>
   * At least one rather than exactly one: the alert is raised for every thread of the
   * server which ends that way, and a thread of some other component ending during the
   * test must not turn this into a failure of the wrong test.
   */
  private static void assertUncaughtExceptionAlertRaisedSince(int initialAlerts, String message)
  {
    Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION))
        .as(message)
        .isGreaterThanOrEqualTo(initialAlerts + 1);
  }

  /**
   * Test case for [Issue 922]: a change whose ack could not be published still takes the
   * road its own replay decided.
   * <p>
   * The ack of a delivery is published in a finally which every road out of the replay runs
   * through, and it is published on the session that delivery came over - which is being
   * torn down when the replay of the change failed. Whether the change was applied is not
   * something that publish can tell, so a throw there is reported and the replay carries on
   * to the road the change itself decided: this one failed, so it is kept out of the
   * ServerState, counted and asked for again.
   * <p>
   * Left to unwind, that throw would step over the give-back which follows it and leave the
   * change owned by a thread which is not replaying it anymore.
   */
  @Test
  public void aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain"));

    assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseAckThrows::new,
        21, "user.922.3", "the ack of this change throws on the way out of the replay");
  }

  /**
   * Test case for [Issue 922] and [Issue 923]: an OutOfMemoryError met where the ack of a
   * delivery is published ends the replay thread, the way one met by the replay itself does.
   * <p>
   * A throw from the ack is caught so that it does not unwind the replay past the give-back
   * of the change and past the hand-out of the changes which were waiting for it. A JVM
   * which has run out of memory is the one exception to that: it is not something to carry
   * on replaying from, so it is left to end this thread - the change is given back on the
   * way out, and the uncaught exception handler of DirectoryThread writes the line and
   * raises the alert #923 is about. Caught like every other throw from there, it would have
   * this thread replay the changes which follow on an exhausted heap, with nothing said
   * anywhere.
   */
  @Test
  public void aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain"));

    final Set<Long> replayThreadsBefore = replayThreadIds();
    final int initialUncaughtAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION);
    assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseAckRunsOutOfMemory::new,
        26, "user.922.10", "the ack of this change runs out of memory");
    /*
     * Waited for rather than read the moment the change lands: the change is given back -
     * and so replayed by another thread - while the thread which met the error is still
     * unwinding, through the session restart and then through the handler which raises the
     * alert. The alert is what the rethrow is for, and the handler raises it once run() has
     * returned, so it lands after the thread is gone and is waited for in the same breath.
     */
    TestTimer timer = new TestTimer.Builder()
      .maxSleep(30, SECONDS)
      .sleepTimes(200, MILLISECONDS)
      .toTimer();
    timer.repeatUntilSuccess(new CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        assertFalse(replayThreadIds().containsAll(replayThreadsBefore),
            "an OutOfMemoryError met where the ack is published must end the replay thread"
                + " which met it");
        assertUncaughtExceptionAlertRaisedSince(initialUncaughtAlerts,
            "the replay thread an OutOfMemoryError met where the ack is published ended must"
                + " have raised the alert #923 asks for on its way out");
      }
    });
  }

  /**
   * Test case for [Issue 922] and [Issue 923]: a change whose replay is unwound once the
   * ack of its delivery is out is given back, and the thread which was replaying it stays.
   * <p>
   * The catches of the replay itself span the roads which decide what became of the change,
   * and the ack is published once that is decided. What the replay runs afterwards - the
   * give-back of a change which failed, and the hand-out of the changes which were waiting
   * for it - is past every one of them: a throw there unwinds {@code replay()} with the
   * change still owned by this thread, and a change owned by a thread which is not replaying
   * it anymore is refused as a duplicate on every later delivery. So it is given back on the
   * way out, and the error is left to the replay thread, whose catch is what keeps it alive
   * for the changes which follow (issue #923).
   */
  @Test
  public void aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain"));

    final Set<Long> replayThreadsBefore = replayThreadIds();
    assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseReplayIsUnwoundAfterItsAck::new,
        24, "user.922.6", "the replay of this change is unwound once its ack is out");
    Assertions.assertThat(replayThreadIds())
        .as("an Error which unwinds a replay must not end the thread which met it (issue #923)")
        .containsAll(replayThreadsBefore);
  }

  /**
   * Test case for [Issue 922]: a change whose replay keeps being unwound after its ack is
   * given up on rather than asked for forever.
   * <p>
   * The change is given back on the road a failed replay takes rather than handed back bare,
   * so the failure counts against the give-up budget of the change: a change which keeps
   * unwinding the replays it is given to is eventually recorded as one this replica could
   * not apply, and the administrator is told that it now diverges. Handed back bare it would
   * be asked for, and have this domain restart its session for it, for as long as the server
   * is up - which is the wedge this issue is about wearing another face.
   */
  @Test
  public void aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn() throws Exception
  {
    testSetUp("aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn"));

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

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
    domain.resetUnreplayedChangeAlertThrottle();
    final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
    setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
    try
    {
      final CSN csn = new CSNGenerator(25, TimeThread.getTime()).newCSN();
      final List<Modification> mods =
          generatemods("description", "the replay of this change is unwound after its ack");

      /*
       * Nothing sends this change again - it never travelled a session - so every delivery
       * of it is made here, until this replica gives up on the change whose replay it can
       * not run to its end.
       */
      TestTimer timer = new TestTimer.Builder()
        .maxSleep(120, SECONDS)
        .sleepTimes(200, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          if (!domain.getServerState().cover(csn))
          {
            domain.processUpdate(new ModifyMsgWhoseReplayIsUnwoundAfterItsAck(csn, dn, mods, uuid));
          }
          assertTrue(domain.getServerState().cover(csn),
              "a change whose replay keeps being unwound must be given up on");
        }
      });
      assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
          "the change which was given up on must be counted as failed, once");
      Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
          .as("the administrator must be told that this replica now diverges")
          .isGreaterThan(initialAlerts);
    }
    finally
    {
      resetReplayGiveUpDelay();
    }
  }

  /**
   * Test case for [Issue 922]: the changes parked behind a change whose ack could not be
   * published are replayed rather than left waiting for a thread which is gone.
   * <p>
   * A change which is waiting for another one is handed out by {@code getNextUpdate()},
   * which the replay runs once it is done with the change it was given - after the ack of
   * that delivery has been published. A throw from that ack used to unwind the replay past
   * it, and the change had been committed by then, so nothing was owed back and nothing was
   * handed out: the changes parked behind it stayed parked, and the ServerState of this
   * domain stayed behind them until some other change was replayed here.
   * <p>
   * What tells the two apart is which thread replays the parked change. It is handed to
   * whichever thread cleared the change it was waiting for, so it is replayed by the very
   * thread which has just committed the change whose ack threw - a change nobody handed out
   * is replayed by no one at all, and the wait below is what says so.
   */
  @Test
  public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
  {
    testSetUp("theChangesParkedBehindAChangeWhoseAckFailedAreReplayed");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : theChangesParkedBehindAChangeWhoseAckFailedAreReplayed"));

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
    final String parentUUID = "26262626-2626-2626-2626-262626262626";
    final String childUUID = "27272727-2727-2727-2727-272727272727";
    final Entry parent = TestCaseUtils.makeEntry(
        "dn: ou=parked.922," + baseDN,
        "objectClass: top",
        "objectClass: organizationalUnit",
        "ou: parked.922",
        "entryUUID: " + parentUUID);
    final Entry child = TestCaseUtils.makeEntry(
        "dn: uid=user.922.8,ou=parked.922," + baseDN,
        "objectClass: top",
        "objectClass: person",
        "objectClass: organizationalPerson",
        "objectClass: inetOrgPerson",
        "uid: user.922.8",
        "cn: Aaccf Amar",
        "sn: Amar",
        "entryUUID: " + childUUID);

    /*
     * Both adds are held at the pre-parse plugin point, one at a time. The first park is
     * what keeps the parent listed as pending - and owned by the thread replaying it - while
     * the child is checked for dependencies, so the child is parked behind a change which is
     * in flight rather than behind one which has already been applied.
     */
    final CSN parentCsn = gen.newCSN();
    final CSN childCsn = gen.newCSN();
    final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
        OperationType.ADD, "PreParse",
        op -> parentCsn.equals(OperationContext.getCSN(op))
            || childCsn.equals(OperationContext.getCSN(op)));
    try
    {
      domain.processUpdate(new AddMsgWhoseAckThrows(parentCsn, parent.getName(), parentUUID,
          baseUUID, parent.getObjectClassAttribute(), parent.getAllAttributes()));
      final Thread replayingParent = parked.awaitParked(60, SECONDS);

      domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
          child.getObjectClassAttribute(), child.getAllAttributes(), null));

      /*
       * The parent is applied and its ack throws where it is published. The replay carries
       * on all the same, and the child is the change it hands itself next.
       */
      parked.release();
      final Thread replayingChild = parked.awaitParked(60, SECONDS);
      Assertions.assertThat(replayingChild)
          .as("the change which was parked must be replayed by the thread which cleared what"
              + " it was waiting for, rather than be left waiting")
          .isSameAs(replayingParent);
      parked.release();

      assertNotNull(getEntry(child.getName(), 30000, true),
          "the change which was parked behind the one whose ack threw must be applied");
    }
    finally
    {
      parked.deregister();
    }
  }

  /**
   * Test case for [Issue 922]: a change handed out as a dependency is given back when the
   * replay it was handed to is unwound.
   * <p>
   * A change which was parked behind another one is handed out by {@code getNextUpdate()}
   * to the thread which cleared what it was waiting for, and that thread owns it from then
   * on. The give-back on the way out of an unwound replay asks which change this thread
   * owns, so the hand-out has to be recorded where that question is answered, not only on
   * the change: left out, the change would stay owned by a thread which is not replaying
   * it anymore, and every later delivery of it would be refused as a duplicate - the wedge
   * of this issue, on the dependency road.
   * <p>
   * The parent is held at the pre-parse plugin point while the child is delivered, so the
   * child is parked behind a change in flight, and the thread which is thrown out of the
   * child is read at the same plugin point: it must be the one which committed the parent,
   * which is what says the child was handed out rather than taken off the queue. The child
   * is thrown out of once, inside the replay, and then unwound past its ack, on the road
   * every catch of the replay has already run on.
   */
  @Test
  public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() throws Exception
  {
    testSetUp("aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound"));

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    final CSNGenerator gen = new CSNGenerator(29, TimeThread.getTime());
    final String parentUUID = "29292929-2929-2929-2929-292929292929";
    final String childUUID = "30303030-3030-3030-3030-303030303030";
    final Entry parent = TestCaseUtils.makeEntry(
        "dn: ou=handed-out.922," + baseDN,
        "objectClass: top",
        "objectClass: organizationalUnit",
        "ou: handed-out.922",
        "entryUUID: " + parentUUID);
    final Entry child = TestCaseUtils.makeEntry(
        "dn: uid=user.922.11,ou=handed-out.922," + baseDN,
        "objectClass: top",
        "objectClass: person",
        "objectClass: organizationalPerson",
        "objectClass: inetOrgPerson",
        "uid: user.922.11",
        "cn: Aaccf Amar",
        "sn: Amar",
        "entryUUID: " + childUUID);

    final CSN parentCsn = gen.newCSN();
    final CSN childCsn = gen.newCSN();
    final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
    final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
        OperationType.ADD, "PreParse", op -> parentCsn.equals(OperationContext.getCSN(op)));
    /*
     * The thread which reaches the plugin point with the child is the one replaying it, and
     * it is read there rather than parked: a park would hold the child where the throw is
     * made, and it is the throw which is wanted.
     */
    final AtomicReference<Thread> replayingChild = new AtomicReference<>();
    final ThrownFromReplay thrown = ShortCircuitPlugin.throwFromReplayedOperations(
        OperationType.ADD, "PreParse",
        op ->
        {
          if (!childCsn.equals(OperationContext.getCSN(op)))
          {
            return false;
          }
          replayingChild.set(Thread.currentThread());
          return true;
        },
        () -> new LinkageError("the replay of the change which was handed out meets an Error"),
        1);
    try
    {
      domain.processUpdate(new AddMsg(parentCsn, parent.getName(), parentUUID, baseUUID,
          parent.getObjectClassAttribute(), parent.getAllAttributes(), null));
      final Thread replayingParent = parked.awaitParked(60, SECONDS);

      domain.processUpdate(new AddMsgWhoseReplayIsUnwoundAfterItsAck(childCsn, child.getName(),
          childUUID, parentUUID, child.getObjectClassAttribute(), child.getAllAttributes()));

      // The parent is applied, and the child is the change its thread hands itself next.
      parked.release();

      TestTimer timer = new TestTimer.Builder()
        .maxSleep(60, SECONDS)
        .sleepTimes(200, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          assertEquals(thrown.thrownCount(), 1,
              "the change which was handed out must have been thrown out of");
        }
      });
      Assertions.assertThat(replayingChild.get())
          .as("the change which was parked must be replayed by the thread which cleared what"
              + " it was waiting for: that is the hand-out this test is about")
          .isSameAs(replayingParent);

      /*
       * The child was thrown out of and its replay was then unwound, so it is not in the
       * data and must not be in the ServerState - and it must not be given up on either: it
       * is asked for again. The delivery which asks for it is made here, the way
       * assertChangeIsDeliveredAgainAfter() makes it: nothing sends the change again, since
       * it never travelled a session, and a delivery is dropped rather than queued while the
       * session is being restarted. A delivery of a change a thread still owns is refused as
       * the duplicate it is, which is where a change left owned by a thread which is not
       * replaying it never comes back.
       */
      assertFalse(domain.getServerState().cover(childCsn),
          "a change whose replay was unwound must be asked for again, not recorded as replayed");
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          if (!domain.getServerState().cover(childCsn))
          {
            domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
                child.getObjectClassAttribute(), child.getAllAttributes(), null));
          }
          assertTrue(domain.getServerState().cover(childCsn),
              "the change must be recorded as replayed once it has been delivered again");
        }
      });
      assertNotNull(getEntry(child.getName(), 30000, true),
          "the change which was handed out must be applied by the delivery which took over"
              + " from the one which was unwound");
      assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
          "a change which was delivered again must not be counted as one this replica gave up on");
      assertTrue(replayingParent.isAlive(),
          "an Error which unwinds a replay must not end the thread which met it (issue #923)");
    }
    finally
    {
      thrown.deregister();
      parked.deregister();
    }
  }

  /**
   * Test case for [Issue 922]: the ack of a delivery whose replay threw an Error says that
   * the change was not applied.
   * <p>
   * An assured write in SAFE_READ mode is told that its change is durable here by the ack
   * this replica publishes, and it is published in a finally which every road out of the
   * replay runs through - the roads an Error unwinds among them. A replica which is asking
   * for a change again must never have told a master that the change is in its data, so the
   * ack of a delivery whose replay threw reports the error and names this replica.
   * <p>
   * The error is thrown from a plugin point which runs inside the operation, so the change
   * travels a real session and the ack can be read off the broker which published it - the
   * counters can not report it, since handing the change back restarts the session and that
   * resets every one of them.
   */
  @Test
  public void theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt() throws Exception
  {
    testSetUp("theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt"));

    final int serverId = 28;
    /*
     * In the group of the replication server, so that what is 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.922.9," + baseDN,
          "objectClass: top",
          "objectClass: person",
          "objectClass: organizationalPerson",
          "objectClass: inetOrgPerson",
          "uid: user.922.9",
          "cn: Aaccf Amar",
          "sn: Amar");
      final String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
      final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);

      final CSN csn = gen.newCSN();
      /*
       * Thrown out of one replay and no more: the delivery which takes over from the one
       * which was unwound is what applies the change, and a change this replica could never
       * replay would be given up on rather than acknowledged twice.
       */
      final ThrownFromReplay thrown = ShortCircuitPlugin.throwFromReplayedOperations(
          OperationType.DELETE, "PreParse",
          op -> csn.equals(OperationContext.getCSN(op)),
          () -> new LinkageError("the replay of this change meets an Error"), 1);
      try
      {
        final DeleteMsg delete = new DeleteMsg(tmp.getName(), csn, uuid);
        delete.setAssured(true);
        delete.setAssuredMode(AssuredMode.SAFE_READ_MODE);
        broker.publish(delete);

        final AckMsg ack = awaitAck(broker, csn);
        assertTrue(ack.hasReplayError(),
            "the ack of a delivery whose replay threw an Error must report it rather than be"
                + " the plain ack a master would take for a durable write");
        assertFalse(ack.hasTimeout(),
            "the ack must be the one the delivery published, not the one the replication"
                + " server makes up when it gives up waiting for it");
        Assertions.assertThat(ack.getFailedServers())
            .as("the replica whose replay threw must be the one the ack names")
            .containsExactly(domainSid);
        assertEquals(thrown.thrownCount(), 1, "the replay of the change must have been unwound");
      }
      finally
      {
        thrown.deregister();
      }

      /*
       * The change was given back rather than recorded as replayed, so the delivery which
       * follows applies it - which is what the ack above said had not happened yet.
       */
      assertNull(getEntry(tmp.getName(), 30000, false),
          "the change must be applied by the delivery which took over from the one whose"
              + " replay threw");
      assertTrue(domain.getServerState().cover(csn),
          "the change must be recorded as replayed once it has been applied");
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Test case for [Issue 922]: a change whose replay keeps throwing an Error is given up
   * on rather than asked for forever.
   * <p>
   * An Error takes the road every other failed replay takes, so the failures it leaves
   * behind count against the give-up budget of the change: a change this replica can never
   * apply must not hold its ServerState - and every change which follows it, from every
   * master - back for good, whether its replay reported the failure or threw it.
   */
  @Test
  public void aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn() throws Exception
  {
    testSetUp("aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn"));

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

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
    domain.resetUnreplayedChangeAlertThrottle();
    final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
    setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
    try
    {
      final CSN csn = new CSNGenerator(22, TimeThread.getTime()).newCSN();
      final List<Modification> mods =
          generatemods("description", "the replay of this change keeps throwing");

      /*
       * Nothing sends this change again - it never travelled a session - so every delivery
       * of it is made here, until this replica gives up on the change it can not replay.
       */
      TestTimer timer = new TestTimer.Builder()
        .maxSleep(120, SECONDS)
        .sleepTimes(200, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          if (!domain.getServerState().cover(csn))
          {
            domain.processUpdate(new ModifyMsgWhoseReplayThrows(csn, dn, mods, uuid,
                () -> new LinkageError("the replay of this change meets an Error")));
          }
          assertTrue(domain.getServerState().cover(csn),
              "a change whose replay keeps throwing must be given up on");
        }
      });
      assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
          "the change which was given up on must be counted as failed, once");
      Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
          .as("the administrator must be told that this replica now diverges")
          .isGreaterThan(initialAlerts);
    }
    finally
    {
      resetReplayGiveUpDelay();
    }
  }

  /** A delivery of a change whose replay does not run to its end. */
  private interface FailingDelivery
  {
    ModifyMsg newDelivery(CSN csn, DN dn, List<Modification> mods, String entryUUID);
  }

  /**
   * Delivers a change whose replay does not run to its end, then checks that the change is
   * neither recorded as replayed nor lost: the delivery which follows must be able to
   * replay it.
   *
   * @param delivery the delivery whose replay is to be unwound
   * @param serverId the replica the change comes from, one per test so that the CSNs of
   *                 one are never covered by the ServerState another left behind
   * @param uid the entry the change is made on
   * @param description the value the change writes, once it is replayed
   */
  private void assertChangeIsDeliveredAgainAfter(
      FailingDelivery delivery, int serverId, String uid, final String description) throws Exception
  {
    Entry tmp = TestCaseUtils.addEntry(
        "dn: uid=" + uid + "," + baseDN,
        "objectClass: top",
        "objectClass: person",
        "objectClass: organizationalPerson",
        "objectClass: inetOrgPerson",
        "uid: " + uid,
        "cn: Aaccf Amar",
        "sn: Amar");
    final DN dn = tmp.getName();
@@ -2601,12 +3330,10 @@
    domain.resetUnreplayedChangeAlertThrottle();
    final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);

    final CSNGenerator gen = new CSNGenerator(18, TimeThread.getTime());
    final CSN csn = gen.newCSN();
    final String description = "the replay must fail once the operation is built";
    final CSN csn = new CSNGenerator(serverId, TimeThread.getTime()).newCSN();
    final List<Modification> mods = generatemods("description", description);

    domain.processUpdate(new ModifyMsgWhoseOperationRefusesAControl(csn, dn, mods, uuid));
    domain.processUpdate(delivery.newDelivery(csn, dn, mods, uuid));

    /*
     * Long enough to outlast the session restart the failure asks for: a change which is
@@ -2615,24 +3342,22 @@
    for (int i = 0; i < MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY; i++)
    {
      assertFalse(domain.getServerState().cover(csn),
          "a change whose operation was built must be asked for again, not recorded as replayed");
          "a change whose replay was unwound must be asked for again, not recorded as replayed");
      Thread.sleep(200);
    }
    assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
        "a change which is still to be delivered again must not be counted as given up on");
    assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
        "a change which is still to be delivered again must not be alerted on as a divergence");
    assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE),
        initialAlerts, "a change which is still to be delivered again must not be alerted on"
            + " as a divergence");

    /*
     * The failed change is the barrier which holds this domain's ServerState back until
     * it is replayed, and the replication server sending it again is what replays it.
     * Nothing sends this one - it never travelled a session - so the delivery which takes
     * over from the one which failed is made here, and it is made until it is taken: a
     * delivery is dropped rather than queued while the listener thread is down, which it
     * is for as long as the recovery is restarting the session, and the monitor entry
     * read above comes back with the broker rather than with the listener. A delivery of
     * a change a replay thread owns is refused as the duplicate it is, and the ServerState
     * keeps this from delivering a change which was replayed a second time.
     * Nothing sends this change again - it never travelled a session - so the delivery
     * which takes over from the one which was unwound is made here, and it is made until
     * it is taken: a delivery is dropped rather than queued while the listener thread is
     * down, which it is for as long as the recovery is restarting the session. A delivery
     * of a change a replay thread owns is refused as the duplicate it is, so this is where
     * a change left owned by a thread which is gone never comes back.
     */
    TestTimer timer = new TestTimer.Builder()
      .maxSleep(60, SECONDS)
@@ -2652,7 +3377,83 @@
      }
    });
    checkEntryHasAttributeValue(dn, "description", description, 30,
        "the change must be applied by the delivery which took over from the failed one");
        "the change must be applied by the delivery which took over from the one which failed");
  }

  /**
   * Returns the identities of the replay threads which are running.
   * <p>
   * Read by identity rather than counted where a test is about one thread in particular
   * having ended: the pool belongs to the server rather than to a test, so a count says
   * whether it is the size it was, not whether the thread which met the error is the one
   * which is gone.
   */
  private static Set<Long> replayThreadIds()
  {
    final Set<Long> running = new HashSet<>();
    for (Thread thread : Thread.getAllStackTraces().keySet())
    {
      if (thread.isAlive() && thread.getName().startsWith("Replica replay thread "))
      {
        running.add(thread.getId());
      }
    }
    return running;
  }

  /**
   * A ModifyMsg whose replay throws an Error.
   * <p>
   * It is thrown where the operation is built, which is inside the replay and past the
   * point where the change was marked as being replayed by the thread which took it: what
   * this pins is the road out of a replay which no {@code catch} of the replay itself used
   * to run on.
   */
  private static final class ModifyMsgWhoseReplayThrows extends ModifyMsg
  {
    private final Supplier<Error> error;

    private ModifyMsgWhoseReplayThrows(
        CSN csn, DN dn, List<Modification> mods, String entryUUID, Supplier<Error> error)
    {
      super(csn, dn, mods, entryUUID);
      this.error = error;
    }

    @Override
    public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
    {
      throw error.get();
    }
  }

  /**
   * A ModifyMsg whose replay fails and whose ack throws on the way out of it.
   * <p>
   * Its operation is built and can not be prepared for its replay, the way
   * {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
   * change owned by the replay thread. The ack of the delivery is then published in the
   * finally every road out of the replay runs through, and this one throws there - which is
   * what a session being torn down does.
   */
  private static final class ModifyMsgWhoseAckThrows
      extends ModifyMsgWhoseOperationRefusesAControl
  {
    private ModifyMsgWhoseAckThrows(CSN csn, DN dn, List<Modification> mods, String entryUUID)
    {
      super(csn, dn, mods, entryUUID);
    }

    @Override
    public boolean isAssured()
    {
      /*
       * An Error rather than the exception a session being torn down raises: the two take
       * the same road, and an Error is what a catch of Exception would let past - the guard
       * around the ack has to hold whatever publishing it threw.
       */
      throw new LinkageError("the ack of this delivery can not be published");
    }
  }

  /**
@@ -2725,6 +3526,96 @@
  }

  /**
   * A ModifyMsg whose ack runs out of memory on the way out of a replay which failed.
   * <p>
   * The replay fails first - its operation can not be prepared for the replay, the way
   * {@code ModifyMsgWhoseOperationRefusesAControl} has it fail - so the change is one this
   * replica asks for again, and the ack which says so is where the JVM runs out of memory.
   * That is the one throw from there which is not caught: it ends the replay thread, and
   * the change is given back on the way out.
   */
  private static final class ModifyMsgWhoseAckRunsOutOfMemory
      extends ModifyMsgWhoseOperationRefusesAControl
  {
    private ModifyMsgWhoseAckRunsOutOfMemory(
        CSN csn, DN dn, List<Modification> mods, String entryUUID)
    {
      super(csn, dn, mods, entryUUID);
    }

    @Override
    public boolean isAssured()
    {
      // Read first thing by processUpdateDone(), which is what publishes the ack.
      throw new OutOfMemoryError("the ack of this delivery runs out of memory");
    }
  }

  /**
   * An AddMsg whose ack throws on the way out of a replay which applied it.
   * <p>
   * The change is committed before the ack of its delivery is published, so this is the
   * road on which nothing is owed back to the replication server - and on which the changes
   * parked behind this one are still waiting to be handed out.
   */
  private static final class AddMsgWhoseAckThrows extends AddMsg
  {
    private AddMsgWhoseAckThrows(CSN csn, DN dn, String entryUUID, String parentEntryUUID,
        Attribute objectClasses, Iterable<Attribute> userAttributes)
    {
      super(csn, dn, entryUUID, parentEntryUUID, objectClasses, userAttributes, null);
    }

    @Override
    public boolean isAssured()
    {
      // Read first thing by processUpdateDone(), which is what publishes the ack.
      throw new LinkageError("the ack of this delivery can not be published");
    }
  }

  /**
   * An AddMsg whose replay is unwound once the ack of its delivery is out, the way
   * {@link ModifyMsgWhoseReplayIsUnwoundAfterItsAck} is.
   * <p>
   * What has its replay fail is not the message but the test which delivers it, through a
   * throw at the pre-parse plugin point: an add which is parked behind its parent has to be
   * one whose operation builds and runs, or it would be given up on where no operation could
   * be built from it. The throw is caught inside the replay, so it is what the replay runs
   * once the ack is out - the give-back of the failed change - which reads the CSN off this
   * message and is unwound by it.
   */
  private static final class AddMsgWhoseReplayIsUnwoundAfterItsAck extends AddMsg
  {
    private volatile boolean ackPublished;

    private AddMsgWhoseReplayIsUnwoundAfterItsAck(CSN csn, DN dn, String entryUUID,
        String parentEntryUUID, Attribute objectClasses, Iterable<Attribute> userAttributes)
    {
      super(csn, dn, entryUUID, parentEntryUUID, objectClasses, userAttributes, null);
    }

    @Override
    public boolean isAssured()
    {
      // Read first thing by processUpdateDone(), and by nothing on the way in: a message
      // handed to the domain rather than published is not one this server acknowledges.
      ackPublished = true;
      return super.isAssured();
    }

    @Override
    public CSN getCSN()
    {
      if (ackPublished)
      {
        throw new LinkageError("the replay of this change is unwound once its ack is out");
      }
      return super.getCSN();
    }
  }

  /**
   * Test case for [Issue 908]: a domain being disabled - for an LDIF import, a restore, or
   * a backend being taken offline - must not save its ServerState while a replay thread is
   * half way through applying one of its changes.
@@ -3057,6 +3948,52 @@
  }

  /**
   * A ModifyMsg whose replay is unwound once the ack of its delivery is out.
   * <p>
   * Its operation is built and can not be prepared for its replay, the way
   * {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
   * change owned by the replay thread. The CSN of the change is then read again - by the
   * road which gives it back and asks for it again - and it is that read which throws here:
   * past the ack, past the finally it is published in, and past every catch the replay
   * itself has. So the only thing left to give the change back is the road out of
   * {@code replay()}.
   */
  private static final class ModifyMsgWhoseReplayIsUnwoundAfterItsAck
      extends ModifyMsgWhoseOperationRefusesAControl
  {
    private volatile boolean ackPublished;

    private ModifyMsgWhoseReplayIsUnwoundAfterItsAck(
        CSN csn, DN dn, List<Modification> mods, String entryUUID)
    {
      super(csn, dn, mods, entryUUID);
    }

    @Override
    public boolean isAssured()
    {
      /*
       * processUpdateDone() reads this first and reads nothing else of a delivery which is
       * not assured, so the ack of this one is out by the time it returns. Nothing on the
       * way in reads it: a message handed to the domain rather than published is not one
       * this server acknowledges to anybody.
       */
      ackPublished = true;
      return super.isAssured();
    }

    @Override
    public CSN getCSN()
    {
      if (ackPublished)
      {
        throw new LinkageError("the replay of this change is unwound once its ack is out");
      }
      return super.getCSN();
    }
  }

  /**
   * A ModifyMsg whose operation can not be prepared for its replay.
   * <p>
   * The operation is built - so the replay is past the point where a message is given up
@@ -3066,7 +4003,7 @@
   * protocol: {@code ModifyMsg.createOperation()} builds an operation whose controls can
   * be added to, so this one is handed to the domain rather than published.
   */
  private static final class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
  private static class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
  {
    private ModifyMsgWhoseOperationRefusesAControl(
        CSN csn, DN dn, List<Modification> mods, String entryUUID)

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

@@ -17,6 +17,11 @@

import static org.testng.Assert.*;

import java.util.NoSuchElementException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.TestCaseUtils;
@@ -24,6 +29,8 @@
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

@@ -469,6 +476,371 @@
        "a disabled domain forgot the change, so nothing is failing here anymore");
  }

  /**
   * A change is given back by the replay thread which owns it and by nobody else.
   * <p>
   * The release is what lets the next delivery of a change be replayed, so a release
   * which arrives from a thread which does not own the change hands a change which is
   * being replayed right now to a second thread - the double replay the ownership is
   * there to prevent (OPENDJ-1115). It happens when a thread reports a failure on a
   * change which has been taken over since, which is every road out of a replay that is
   * not the one which failed: an Error unwinding the replay thread, or an exception on
   * the way to the ack (issue #922).
   */
  @Test
  public void replayFailedIsIgnoredForAThreadWhichDoesNotOwnTheChange() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg delivery = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(delivery));
    assertTrue(pendingChanges.markInProgress(delivery));

    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        pendingChanges.replayFailed(csn);
      }
    });

    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
        "a change another thread is replaying must not be taken over");
    assertEquals(pendingChanges.getQueueSize(), 1);
    assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), csn,
        "a release which was ignored must leave the change with the thread which owns it");

    // The thread which owns the change is still the one which decides its fate.
    pendingChanges.replayFailed(csn);
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
        "the change its owner gave back must be taken over by the next delivery");
    /*
     * The give-back on the way out of an unwound replay reads this: a thread which gave a
     * change back and is then unwound - the session restart it runs next can throw - must
     * find nothing to give back, or it would count a second failure against a change it
     * does not own anymore.
     */
    assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
        "a change which was given back is not one this thread gives back again");
  }

  /**
   * A change which was parked because it depends on another one is handed to whichever
   * replay thread clears the change it was waiting for, rather than replayed by the
   * thread which parked it. The thread it is handed to is the one which owns it from
   * then on: the failure of the replay it is about to be given is reported by that
   * thread, and a give-back which comes from a thread the change was never handed to is
   * ignored (issue #922).
   * <p>
   * The hand-out is what the give-back on the way out of an unwound replay must read as
   * well as the owner: a change handed out by {@code getNextUpdate()} whose replay is then
   * unwound would otherwise be left owned by a thread which is not replaying it anymore,
   * with every later delivery of it refused as a duplicate - the wedge of that issue, on
   * the dependency road.
   */
  @Test
  public void aChangeTakenAsADependencyIsOwnedByTheThreadWhichTakesIt() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN deleted = generator.newCSN();
    final CSN renamed = generator.newCSN();

    final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(delete));
    assertTrue(pendingChanges.markInProgress(delete));

    // A rename into the DN that delete is on: it can only be replayed once the delete has been.
    final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
    assertTrue(pendingChanges.putRemoteUpdate(rename));
    assertTrue(pendingChanges.markInProgress(rename));
    assertTrue(pendingChanges.checkDependencies(rename),
        "the rename must wait for the delete of the entry it renames into");

    // The delete has been replayed, so the rename is handed to the thread which replayed it.
    pendingChanges.commit(deleted);

    final AtomicReference<LDAPUpdateMsg> taken = new AtomicReference<>();
    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        taken.set(pendingChanges.getNextUpdate());
        assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), renamed,
            "the change handed out by getNextUpdate() must be the one the thread it was"
                + " handed to gives back on the way out of an unwound replay");
        // ... and the replay it was taken for failed.
        pendingChanges.replayFailed(renamed);
        assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
            "a change which was given back is not one this thread gives back again");
      }
    });

    assertSame(taken.get(), rename, "the change which was waiting must be handed out");
    assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
        "the change the thread which took it gave back must be taken over by the next delivery");
  }

  /**
   * A thread which reports on a change it gave back a moment ago must not record it as
   * replayed: the delivery which took the change over is being applied right now, so the
   * ServerState would move past a change which is not in the data yet (issue #889), and
   * the thread which is applying it would find nothing left to commit.
   * <p>
   * The give-back on the way out of an unwound replay is what makes this reachable: it
   * runs wherever the replay was left, so the decision a thread carries and the record it
   * makes of it can be a delivery apart (issue #922).
   */
  @Test
  public void commitIsRefusedForAThreadWhichDoesNotOwnTheChange() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg delivery = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(delivery));
    assertTrue(pendingChanges.markInProgress(delivery));

    // The thread which gave up on an earlier delivery of this change records it as
    // replayed while this delivery is being applied.
    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          pendingChanges.commit(csn);
          fail("a change another replay thread owns must not be recorded as replayed");
        }
        catch (NoSuchElementException expected)
        {
          // There is no change here for that thread to record, which is what its caller
          // reports as ERR_OPERATION_NOT_FOUND_IN_PENDING.
        }
      }
    });

    assertTrue(state.isEmpty(),
        "a change which is being applied must not be recorded in the ServerState");
    assertEquals(pendingChanges.getQueueSize(), 1, "the change must stay listed as pending");

    // The thread which owns the change records it once it really has been applied.
    pendingChanges.commit(csn);

    assertTrue(state.cover(csn));
    assertEquals(pendingChanges.getQueueSize(), 0);
  }

  /**
   * The same holds for the failures which decide when this replica gives up on a change:
   * a thread which does not own the change reports none, so the give-up budget of the
   * delivery which took it over is left alone rather than spent by the one before it
   * (issue #922).
   */
  @Test
  public void recordReplayFailureIsIgnoredForAThreadWhichDoesNotOwnTheChange() throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg delivery = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(delivery));
    assertTrue(pendingChanges.markInProgress(delivery));

    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        assertNull(pendingChanges.recordReplayFailure(csn, 1000),
            "a change another replay thread owns is not this one's to give up on");
      }
    });

    assertEquals(pendingChanges.recordReplayFailure(csn, 2000).getAttempts(), 1,
        "only the failures of the delivery which owns the change must be counted");
  }

  /**
   * The change a thread parked as waiting for another one is not the change it is
   * replaying: it is handed to whichever thread clears what it waits for, so a give-back on
   * the way out of an unwound replay must leave it alone. Releasing it without taking it out
   * of the changes which are waiting would have the same change handed to two threads
   * (issue #922).
   * <p>
   * The deliveries are taken in the order a replay thread takes them: one at a time, off
   * the queue the pool shares. So the change which is parked here is parked by the thread
   * which was replaying it, and the change that thread is replaying afterwards is the next
   * delivery it took - never a second one it holds at the same time.
   */
  @Test
  public void theChangesParkedAsDependenciesAreNotOwnedByTheThreadWhichParkedThem()
      throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN deleted = generator.newCSN();
    final CSN renamed = generator.newCSN();
    final CSN taken = generator.newCSN();

    /*
     * The delete is being replayed by another thread of the pool, which is what a change
     * waits for: the thread which parks a change is not the one applying the change it
     * waits for.
     */
    final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(delete));
    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        assertTrue(pendingChanges.markInProgress(delete),
            "the delete must be listed as being replayed by the thread which took it");
      }
    });

    // A rename into the DN that delete is on, parked by this very thread.
    final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
    assertTrue(pendingChanges.putRemoteUpdate(rename));
    assertTrue(pendingChanges.markInProgress(rename));
    assertTrue(pendingChanges.checkDependencies(rename),
        "the rename must wait for the delete of the entry it renames into");

    assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
        "a change this thread parked as waiting for another one is not one it gives back:"
            + " it is handed to whichever thread clears what it waits for");

    // The delivery this thread took once the change it parked was out of its hands.
    final DeleteMsg next = deleteMsg(taken, "uuid-3");
    assertTrue(pendingChanges.putRemoteUpdate(next));
    assertTrue(pendingChanges.markInProgress(next));

    assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), taken,
        "the change this thread is replaying is the one it must give back, not the one it"
            + " parked as waiting for another");
  }

  /**
   * A change which is in the data is not one to give back either, and it stays listed for as
   * long as an older change holds the ServerState back: a give-back on the way out of an
   * unwound replay must not ask for a change this replica has already applied (issue #922).
   * <p>
   * What this pins is that behaviour rather than one of the two conditions which hold it:
   * {@code commit()} marks the change and clears its owner in the same write-locked step, so
   * a committed change is never an owned one and the {@code isCommitted} arm of
   * {@code getChangeOwnedByCurrentThread()} can not be told from the owner check by any test.
   */
  @Test
  public void aChangeWhichWasAppliedIsNotOwnedAnymore() throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN held = generator.newCSN();
    final CSN applied = generator.newCSN();

    // An older change nobody has replayed holds the ServerState back, so the change this
    // thread applies stays listed here once it has been committed.
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(held, "uuid-1")));
    final DeleteMsg delivery = deleteMsg(applied, "uuid-2");
    assertTrue(pendingChanges.putRemoteUpdate(delivery));
    assertTrue(pendingChanges.markInProgress(delivery));

    assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), applied,
        "the change this thread is replaying must be the one it owns");

    pendingChanges.commit(applied);

    assertEquals(pendingChanges.getQueueSize(), 2,
        "the change which was applied is held back by the one before it");
    assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
        "a change which is in the data must not be given back on the way out of a replay");
  }

  /**
   * A change which was waiting is handed out once: the thread it is handed to owns it from
   * then on, and the threads which ask next are told there is nothing to take. Two threads
   * handed the same change would replay it twice, which is what the ownership is there to
   * prevent (OPENDJ-1115).
   */
  @Test
  public void aChangeWhichWasWaitingIsHandedOutOnce() throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN deleted = generator.newCSN();
    final CSN renamed = generator.newCSN();

    final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(delete));
    assertTrue(pendingChanges.markInProgress(delete));

    final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
    assertTrue(pendingChanges.putRemoteUpdate(rename));
    assertTrue(pendingChanges.markInProgress(rename));
    assertTrue(pendingChanges.checkDependencies(rename),
        "the rename must wait for the delete of the entry it renames into");

    assertNull(pendingChanges.getNextUpdate(),
        "a change whose dependency still stands must not be handed out");

    pendingChanges.commit(deleted);

    assertSame(pendingChanges.getNextUpdate(), rename,
        "the change which was waiting must be handed to the thread which cleared it");
    assertNull(pendingChanges.getNextUpdate(),
        "a change which has been handed out must not be handed out again");
  }

  /** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */
  private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception
  {
    return new ModifyDNMsg(DN.valueOf("cn=uuid-2,dc=example,dc=com"), csn, "uuid-2",
        null, false, null, "cn=uuid-1");
  }

  /**
   * Runs the provided work on a thread of its own, waits for it and reports what it
   * threw.
   * <p>
   * What the work throws has to be carried back here: an assertion which fails on another
   * thread is lost to a bare {@link Thread#join()}, and every assertion these tests make
   * about what a thread which does not own a change is answered is made on that thread.
   */
  private static void runAndJoin(Runnable runnable) throws Exception
  {
    final FutureTask<Void> task = new FutureTask<>(runnable, null);
    final Thread thread = new Thread(task, "another replay thread");
    thread.start();
    thread.join();
    try
    {
      task.get();
    }
    catch (ExecutionException e)
    {
      // Report what the work threw rather than the wrapper this task put around it: an
      // assertion which failed is an Error, and it is the failure worth reading.
      final Throwable cause = e.getCause();
      if (cause instanceof Error)
      {
        throw (Error) cause;
      }
      if (cause instanceof Exception)
      {
        throw (Exception) cause;
      }
      throw e;
    }
  }

  private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception
  {
    return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID);