[#986] Give back the changes a replay thread the pool stopped had parked (#...

Valery Kharseko

2 days ago 80481f756d71bd58b4bda627758dcd774e9d5dcd

[#986] Give back the changes a replay thread the pool stopped had parked (#988)

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

@@ -2760,20 +2760,35 @@
       * 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.
       * The changes this thread parked as waiting for another change are given back too,
       * and before the road above runs: that road restarts the session, and a change which
       * is still owned when the replication server sends it again over it is turned down as
       * a duplicate - the one delivery which could have taken it over (issue #954).
       *
       * 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.
       * Which change this thread owns is read first of all, 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. It is read before the parked changes
       * are given back rather than after, because that give-back allocates and can throw on
       * the same road, and the last resort below can only hand back a change it was told
       * about.
       */
      CSN owned = null;
      try
      {
        owned = remotePendingChanges.getChangeOwnedByCurrentThread();
        /*
         * Asked for without the backoff on the two roads recoverFromReplayFailure() asks
         * for it without on: a thread which is stopping, and one which an OutOfMemoryError
         * is ending - the backend is not what is going away. What is asked for is not what
         * is run: a request is never answered by less than it asked for, so a restart
         * another road asked for with the backoff, or one given back with it, keeps its
         * wait whichever thread runs it.
         */
        final boolean parkedGivenBack = giveBackParkedChanges(
            replayThreadShutdown.get() || t instanceof OutOfMemoryError
                ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF);
        if (owned != null)
        {
          if (replayThreadShutdown.get() || shutdown.get() || disabled)
@@ -2808,6 +2823,27 @@
            recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError);
          }
        }
        if (parkedGivenBack && !replayThreadShutdown.get() && !sessionHasAnOwner())
        {
          /*
           * The road the change this thread was replaying took may have run the restart the
           * give-back asked for - they ask for the same one - and it may have had none to
           * run: this thread owned no change, or the change it owned was given up on. What
           * is still requested is run here rather than left standing: the state
           * checkpointer would run it within its tick, so this is the latency of the
           * delivery the changes which were handed back wait for, and nothing more - a
           * request this thread leaves is not lost.
           *
           * A thread which is stopping leaves it standing, the way abandonReplay() does:
           * the state checkpointer runs one restart for every change the threads of the
           * pool hand back on their way out, rather than each of them running one while
           * the configuration change which is stopping them waits. A domain whose session
           * has an owner is left alone the way the give-back left it: nothing was asked
           * for on that road, and a request another thread left standing is not this
           * one's to spend on a restart which is refused where it runs.
           */
          runRequestedSessionRestarts();
        }
      }
      catch (Throwable recoveryFailure)
      {
@@ -2819,7 +2855,15 @@
         * 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. A
         * restart which can not run here leaves its request standing, and the state
         * checkpointer of this domain runs it.
         * checkpointer of this domain runs it: one which threw asks for itself again on
         * its way out, and one a domain whose session has an owner would refuse is not
         * run at all rather than spent on the refusal.
         *
         * The restart is asked for once the change is released and not before, the way
         * every road which releases one asks: asked for first, it could be taken and run
         * by another thread while this one still owned the change, and the delivery the
         * new session brought would be turned down as the duplicate of a change a replay
         * thread owns, with nothing left standing to ask for it again.
         */
        if (owned != null)
        {
@@ -2840,21 +2884,43 @@
          {
            suppress(recoveryFailure, reportFailure);
          }
        }
          try
          {
          /*
           * Outside the guard above: two roads reach here with a request standing and no
           * change of this thread's to hand back, and both are the parked changes' - the
           * give-back which released them asks for the restart before it reports them, and
           * a throw out of the report - the JVM which unwound this replay is out of memory
           * - leaves the request standing; and a restart the parked road ran and which
           * threw has asked for one again on its way out. The changes it released are
           * listed, uncommitted and unowned, so the request is what brings them back, and
           * this thread is the one there to run it (issue #954). A give-back which threw
           * before it released anything left the parked changes as they were, owned by this
           * thread and handed out by getNextUpdate() to whichever thread clears what they
           * wait for: nothing here can do better for those.
           *
           * Not run on a domain whose session has an owner, the way no road of a failed
           * replay runs it there: the restart is refused where it runs and the request
           * would be spent on the refusal, while a request left standing is run by the
           * state checkpointer once the owner is gone - or forgotten with the pending
           * changes it was made for, when the owner forgets them on its way out.
           */
          if (!sessionHasAnOwner())
          {
            runRequestedSessionRestarts();
          }
        }
          catch (Throwable restartFailure)
          {
            /*
             * Nothing is left to try here: the change is listed, uncommitted and unowned,
           * Nothing is left to try here: the changes are listed, uncommitted and unowned,
             * and the restart which threw has asked for one again, so the session restart
             * the state checkpointer runs delivers it again. This goes with the throwable
           * the state checkpointer runs delivers them 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);
@@ -3387,12 +3453,14 @@
             * 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.
             * owner, and the index the give-back reads, in the same step - so the change it
             * was replaying is not given back, 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, which hands out the changes parked behind the committed change: the
             * ones this thread parked are given back on the way out of replay() and the
             * session is restarted for them (issue #954), the ones other threads parked 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;
          }
@@ -3968,6 +4036,25 @@
  }

  /**
   * Gives back the changes a replay thread which is stopping parked in this domain.
   * <p>
   * Called by that thread on its way out (issue #986). The restart which brings them back
   * is asked for and left standing, the way the thread leaves the request it makes for the
   * change it abandons: the threads of the pool are stopped one after the other and joined,
   * and each running a restart on its way out would have the configuration change which is
   * stopping them wait for one restart per thread. The state checkpointer of this domain
   * runs one restart for the lot within its tick, and holds it while a total update runs
   * over the session, in either direction - a change delivered again before the pool which
   * replaces this thread is up waits in the replay queue for it. Asked for without the
   * backoff: what went away is a replay thread, not the backend, and these changes were
   * never applied here.
   */
  void giveBackChangesParkedByStoppingThread()
  {
    giveBackParkedChanges(SessionRestart.NOW);
  }

  /**
   * Restarts the session as long as changes which could not be replayed are waiting to be
   * delivered again.
   */
@@ -4076,6 +4163,84 @@
  }

  /**
   * Gives back the changes this replay thread parked as waiting for another change, on the
   * way out of a replay which was unwound.
   * <p>
   * A parked change is handed out again by {@code getNextUpdate()} alone, which every
   * replay loop of this domain runs once it is done with a change: a parked change is
   * replayed by whichever thread clears the change it was waiting for. A thread whose
   * replay was unwound is not on that road anymore - it takes the next delivery off the
   * replay queue - so a change it parked would be left owned by a thread which is not
   * coming back to it, while every redelivery of it is refused as a duplicate. On a domain
   * which then goes quiet that change is where this replica's ServerState, and every change
   * behind it from every master, stops (issue #954).
   * <p>
   * They are handed back without a failure being counted against them: they were never
   * applied here, so the give-up budget which decides when this replica skips a change it
   * can not apply is not this delivery's to spend, the way it is not for a change abandoned
   * by a replay thread which is stopping.
   * <p>
   * The delivery which carried one published no ack - the ack of a parked change is
   * published by the delivery which replays it - so it is counted as processed here, the
   * way a delivery which is dropped rather than replayed is: that count is of the
   * deliveries this replica took off the session, and these are over. The window they hold
   * is not given back either, and does not need to be: the session they came over is about
   * to be restarted, and a session which starts is given its receive window anew.
   * <p>
   * On a domain whose session has an owner - the domain itself, going away, or a total
   * update into it, from the moment it is asked for - they are released and nothing more,
   * the way {@code abandonReplay()} hands a change back on that road (see
   * {@link #sessionHasAnOwner()}): there is no session of this thread's to restart, and
   * the restart it would ask for is refused where it runs. The domain forgets its pending
   * changes on its way down, the import forgets them at its end, and a change released
   * for a total update which never begins stays listed until the next failed replay of
   * this domain restarts the session, which has the replication server send it again. A
   * line which says the replication server sends the change again would not hold on any
   * of these - a server which is shutting down abandons every change in flight, and none
   * of them is delivered again before it is started back.
   * <p>
   * A replay thread which is stopping calls this through
   * {@link #giveBackChangesParkedByStoppingThread()}, for every domain of this server: what
   * it parked would be left owned by a thread which does not exist anymore, and every
   * redelivery of a change a replay thread owns is refused as a duplicate (issue #986). The
   * request it makes here is left standing for the state checkpointer, the way that thread
   * leaves the request it makes for the change it abandons.
   *
   * @param restart what the session restart is asked for as: with the backoff a failing
   *          backend is owed, or without it on a thread which is stopping or which an
   *          OutOfMemoryError is ending
   * @return whether any change was handed back: a change which nobody owns is one only a
   *         new delivery brings back, so the caller runs the restart asked for them - on
   *         a thread which is not stopping, and on a domain whose session has no owner
   */
  private boolean giveBackParkedChanges(SessionRestart restart)
  {
    final List<CSN> parked = remotePendingChanges.releaseParkedChangesOwnedByCurrentThread();
    if (parked.isEmpty())
    {
      return false;
    }
    if (sessionHasAnOwner())
    {
      // The domain owns its session, or a total update does: both forget the pending
      // changes, and neither leaves a session for this thread to restart.
      return true;
    }
    /*
     * Asked for before the changes are reported: a throw out of the report - the JVM which
     * unwound this replay is out of memory - must not lose the restart which is what brings
     * them back.
     */
    sessionRestarts.request(restart);
    for (CSN csn : parked)
    {
      incProcessedUpdates();
      logger.info(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK, csn, getBaseDN());
    }
    return true;
  }

  /**
   * Gives a change back to the replication server when this replay thread stops before it
   * could apply it.
   * <p>

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

@@ -33,6 +33,7 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -797,6 +798,77 @@
  }

  /**
   * Runs an action on every replication domain of this server, each of them getting its
   * turn whatever one of them threw.
   * <p>
   * The replay threads are shared by every domain of this server, so what concerns the
   * pool is done over the domains rather than over the one a replay was last for: a thread
   * which is stopping gives back what it parked in any of them (issue #986). A throw at one
   * domain must not leave the ones after it as they were - with changes owned by a thread
   * which does not exist anymore - so the first failure is thrown once the loop is over,
   * the others suppressed under it where it records suppression: the error a JVM out of
   * memory prepared beforehand does not - it was made without its constructor, so it keeps
   * no list to record them in - and the JVM hands that one out as often as it is asked for
   * one, so two domains can throw the same instance, and a throwable can not suppress
   * itself. Recording a failure under the first allocates the list it goes in, so on the
   * road this loop is for it can be refused in its turn: a failure which can not be
   * recorded is dropped, and the domains after it still get their turn.
   * <p>
   * The iterator over the domains is the one allocation made before the first of them gets
   * its turn: refused, on the way out of a thread an OutOfMemoryError is ending, it leaves
   * every domain as it was, and it has no cheaper form. The action itself is not one: a
   * caller on that road passes an instance it holds rather than one it makes there.
   * <p>
   * Not synchronized, and it must not become so: it is called by a replay thread on its way
   * out, while {@link #stopReplayThreads()} holds the monitor of this class and waits for
   * that thread to end.
   *
   * @param action what is done on each domain; it declares no checked exception, so what
   *          it throws is an Error or a RuntimeException, and that is what is thrown here
   */
  static void forEachDomain(Consumer<LDAPReplicationDomain> action)
  {
    Throwable failure = null;
    for (LDAPReplicationDomain domain : domains.values())
    {
      try
      {
        action.accept(domain);
      }
      catch (Throwable domainFailure)
      {
        if (failure == null)
        {
          failure = domainFailure;
        }
        else if (failure != domainFailure)
        {
          try
          {
            failure.addSuppressed(domainFailure);
          }
          catch (OutOfMemoryError recordRefused)
          {
            /*
             * The list the record goes in could not be allocated: the failure is dropped
             * rather than allowed to end the loop, since the first one is what is reported
             * and the domains after this one are still to be visited.
             */
          }
        }
      }
    }
    if (failure instanceof Error)
    {
      throw (Error) failure;
    }
    if (failure instanceof RuntimeException)
    {
      throw (RuntimeException) failure;
    }
  }

  /**
   * Gets the number of handled domain objects.
   * @return The number of handled domain objects
   */

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

@@ -17,7 +17,11 @@
 */
package org.opends.server.replication.plugin;

import static java.util.Collections.*;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.SortedMap;
import java.util.SortedSet;
@@ -90,8 +94,10 @@
   * 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.
   * applies it, or parks it as waiting for another change - a parked change is not the one
   * this thread is replaying, and giving it back is
   * {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are
   * waiting rather than this index (issue #954).
   * <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)}
@@ -545,9 +551,10 @@
   * 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).
   * change. The parked ones are left out: they are not the change this thread is replaying,
   * and giving one back is more than dropping its owner - it has to be unparked in the same
   * step, or it would be handed out by two roads at once, which is what
   * {@link #releaseParkedChangesOwnedByCurrentThread()} does (issues #922 and #954).
   * <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,
@@ -570,6 +577,95 @@
  }

  /**
   * Gives back the changes the calling thread parked as waiting for another change, and
   * takes them out of the changes which are waiting in the same step.
   * <p>
   * A parked change stays owned by the thread which parked it while that thread goes on
   * to the changes which follow: {@link #getNextUpdate()} is what hands it out again, to
   * whichever replay thread clears the change it was waiting for, and that thread takes it
   * over. A replay which is unwound leaves the thread which parked it without that road -
   * it takes the next delivery off the replay queue instead - so the change would be left
   * owned by a thread which is never coming back to it, and every redelivery of a change a
   * replay thread owns is refused as a duplicate (issue #954).
   * <p>
   * Unparking a change and giving it back is one step, under both locks, so that only one
   * road can hand it out: a change which was released while it is still listed as waiting
   * would be handed to the thread {@link #getNextUpdate()} gives it to and to the thread
   * which takes over the delivery which follows - the double replay the ownership is there
   * to prevent (OPENDJ-1115).
   * <p>
   * The changes stay listed and uncommitted, and stay among the changes the newer ones are
   * checked against, the way a change whose replay failed does: they are not in the data,
   * so they hold this domain's ServerState back and the changes which follow them keep
   * waiting for them.
   * <p>
   * The changes another thread parked are left alone: a change is given back by the thread
   * which owns it and by nobody else (issue #922). That thread may be inside the dependency
   * checks which parked it - they park a change once per dependency it has - so a change
   * released under it would be listed as waiting again a moment later, and handed out while
   * the delivery which took it over is being replayed.
   *
   * @return the CSNs of the changes it gave back, oldest first; empty when this thread owns
   *         no parked change - the changes a thread parked stay its own, whichever replay
   *         parked them, until {@link #getNextUpdate()} hands them to the thread which
   *         cleared what they wait for or they are given back here
   */
  List<CSN> releaseParkedChangesOwnedByCurrentThread()
  {
    final Thread current = Thread.currentThread();
    /*
     * The second lock is taken inside the try of the first: taking a lock which is held
     * by another thread allocates the node this one waits on, and this runs on the road
     * out of a JVM which has just refused an allocation. A throw out of the second lock
     * would otherwise unwind past the first with that one held by a thread which is
     * ending, and every road which lists, commits or gives back a change would wait for
     * it for good.
     */
    pendingChangesWriteLock.lock();
    try
    {
      dependentChangesLock.lock();
      try
      {
        if (dependentChanges.isEmpty())
        {
          // Nothing is waiting, which is the state every replay but a handful leaves behind.
          return emptyList();
        }
        /*
         * Sized for every change which is waiting, so that the one allocation of this
         * method past the locks is made before anything is taken out of the set. The rule
         * getNextUpdate() states for itself holds here: an allocation which fails once a
         * change has been unparked and released - and this runs on the road out of a JVM
         * which has just refused one - would have moved that change out of the hands which
         * hand it out again, with the caller never told that it did.
         */
        final List<CSN> released = new ArrayList<>(dependentChanges.size());
        final Iterator<PendingChange> it = dependentChanges.iterator();
        while (it.hasNext())
        {
          final PendingChange change = it.next();
          if (change.isOwnedBy(current))
          {
            it.remove();
            change.setOwner(null);
            released.add(change.getCSN());
          }
        }
        return released;
      }
      finally
      {
        dependentChangesLock.unlock();
      }
    }
    finally
    {
      pendingChangesWriteLock.unlock();
    }
  }

  /**
   * 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
@@ -672,8 +768,10 @@
       * 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).
       * over rather than leave it to nobody - and the give-back of the change a replay was
       * unwound on leaves it alone (issue #922). What hands a parked change back is
       * releaseParkedChangesOwnedByCurrentThread(), which unparks it in the same step so
       * that the two roads can not hand it out at once (issue #954).
       */
      changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
    }

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

@@ -24,6 +24,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

import org.opends.server.api.DirectoryThread;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -39,6 +40,15 @@
public class ReplayThread extends DirectoryThread
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
  /**
   * The give-back a thread runs on every domain of this server on its way out, held here
   * rather than written where it is run: a method reference is linked, and its instance
   * made, where it is first run, and this one is first run on the way out of a thread -
   * which an OutOfMemoryError may be ending, on the road this give-back is there for. Made
   * when this class is loaded instead, on a thread which can allocate (issue #986).
   */
  private static final Consumer<LDAPReplicationDomain> GIVE_BACK_PARKED_CHANGES =
      LDAPReplicationDomain::giveBackChangesParkedByStoppingThread;

  private final BlockingQueue<UpdateToReplay> updateToReplayQueue;
  private final ReentrantLock switchQueueLock;
@@ -77,6 +87,53 @@
      logger.trace("Replication Replay thread starting.");
    }

    try
    {
      replayUntilStopped();
    }
    finally
    {
      /*
       * The changes this thread parked as waiting for another change are handed out again
       * by getNextUpdate() alone, which every replay loop of a domain runs once it is done
       * with a change: a parked change is replayed by whichever thread clears the change it
       * was waiting for. A thread which is stopping is not on that road anymore, so what it
       * parked would be left owned by a thread which does not exist, while every redelivery
       * of a change a replay thread owns is refused as a duplicate: on a domain which then
       * goes quiet that change is where the ServerState of this replica, and every change
       * behind it from every master, stops (issue #986).
       *
       * Given back by the thread which owns them, so that the rule every road which reads
       * ownership follows holds on this one as well: a change is given back by the thread it
       * was handed to and by nobody else (issue #922). It is also the one place which sees
       * them all - the pool is shared by every domain of this server, while a replay knows
       * only the domain it was replaying for.
       *
       * The session which brings them back is asked for and left standing, in every domain
       * which got something back, and the state checkpointer of each of them runs it within
       * its tick: a thread on its way out is not held for a session - the threads of the
       * pool are stopped one after the other and joined, and each running a restart of its
       * own would have the configuration change which is stopping them wait for one restart
       * per thread - and a change delivered again before the pool which replaces this one
       * is up waits in the replay queue for it. A thread which an OutOfMemoryError is ending
       * gives back here what it parked in the domains it was not replaying for, on the same
       * terms; the change it was replaying, and what it had parked in that same domain, were
       * given back and asked for again on its way out of replay().
       */
      giveBackParkedChanges();
    }
    if (logger.isTraceEnabled())
    {
      logger.trace("Replication Replay thread stopping.");
    }
  }

  /**
   * Takes the deliveries of the domains of this server off the shared replay queue and
   * replays them, until this thread is stopped.
   */
  private void replayUntilStopped()
  {
    while (!shutdown.get())
    {
      try
@@ -145,9 +202,31 @@
        logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(t));
      }
    }
    if (logger.isTraceEnabled())
    {
      logger.trace("Replication Replay thread stopping.");
    }

  /**
   * Gives back the changes this thread parked as waiting for another change, in every
   * domain of this server.
   * <p>
   * A change which is given back stays listed and uncommitted, the way a change whose replay
   * failed does: it is not in the data, so it holds the ServerState of its domain back and
   * the changes which follow it keep waiting for it, until the delivery which takes it over
   * replays it.
   * <p>
   * Every domain gets its turn whatever one of them threw: what can throw here is an
   * allocation, on the way out of a thread an OutOfMemoryError may be ending - the iterator
   * over the domains, before any of them is reached, then for each of them the list of what
   * it released, made before anything is released, or the report of a change once it is,
   * and between two domains the list a second failure is recorded in under the first, which
   * the loop guards on its own - and the domains which follow would otherwise be left with
   * changes owned by a thread which does not exist anymore, the state this give-back is
   * for. A domain which threw past the release has asked for its restart already: the
   * request is made before the report. The first failure is thrown once the loop is over,
   * so that the uncaught exception handler of {@link DirectoryThread} writes the line and
   * raises the alert.
   */
  private void giveBackParkedChanges()
  {
    MultimasterReplication.forEachDomain(GIVE_BACK_PARKED_CHANGES);
  }
}

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

@@ -58,6 +58,21 @@
  private final AtomicReference<SessionRestart> requested =
      new AtomicReference<>(SessionRestart.NONE);

  SessionRestartRequests()
  {
    /*
     * Every request is made on a replay-failure road, where an allocation may be what has
     * just failed, and the first execution of merge() in a JVM allocates: the call site
     * of its lambda and the VarHandle site inside accumulateAndGet() are linked when they
     * are first run, and nothing runs them before a replay fails. Run once here, on a
     * thread which can allocate, so that a request made on the road out of an
     * OutOfMemoryError asks for nothing the JVM has just refused (issue #954). The same
     * goes for what takes a request, which is run on the same roads.
     */
    merge(SessionRestart.NONE);
    take();
  }

  /**
   * Asks this domain to restart its session.
   *

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

@@ -670,6 +670,9 @@
 server after the replay which owned it was unwound: %s. The change has been released without its \
 failure being counted, and a restart of the session is asked for so that the change is delivered \
 again
NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK_318=Change %s in domain "%s" was waiting for another change \
 to be replayed when the replay thread which parked it went away. The change has not been recorded \
 as replayed and is given back to the replication server, which still owns it and sends it 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/replication/UpdateOperationTest.java

@@ -95,6 +95,7 @@
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
import org.opends.server.types.RawModification;
import org.opends.server.util.StaticUtils;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
import org.opends.server.util.TimeThread;
@@ -3583,6 +3584,508 @@
    }
  }

  /**
   * Test case for [Issue 954]: a change parked as waiting for another one is given back
   * when the replay which parked it is unwound, and the session is restarted for it.
   * <p>
   * A change which waits for another one is parked and stays owned by the replay thread
   * which parked it, while that thread goes on to the changes which follow: it is handed
   * out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
   * it is done, so it is replayed by whichever thread clears the change it was waiting for.
   * A replay which is unwound leaves the thread which parked it without that road - it
   * takes the next delivery off the shared queue instead, and never comes back to the
   * change it parked - and every redelivery of a change a replay thread owns is refused as
   * a duplicate. On a domain which then goes quiet that change is where this replica's
   * ServerState, and every change behind it from every master, stops.
   * <p>
   * The replay which is unwound here is one which applied its change: the change it was
   * replaying is committed and owns nothing anymore by the time the give-back on the way
   * out of {@code replay()} runs, so the restart that give-back asks for is the one thing
   * which has the parked change delivered again - a replay which failed would have asked
   * for the same restart on the road of its own change. The parked change travels the
   * replication server, and nothing but a new session brings it back.
   */
  @Test
  public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain"));

    final DN waitedOn = addEntryForChange("user.954.1");
    final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
    final DN other = addEntryForChange("user.954.2");
    final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    domain.resetUnreplayedChangeAlertThrottle();
    final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "no change of this domain is waiting for another one when this test starts");

    // The replica the changes come from: one of its own, so that the CSNs of this test are
    // never covered by the ServerState another one left behind.
    final int serverId = 30;
    final CSNGenerator gen = new CSNGenerator(serverId, TimeThread.getTime());
    final CSN failing = gen.newCSN();
    final CSN parked = gen.newCSN();
    final CSN unwound = gen.newCSN();
    final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
    final String parkedDescription = "the change which was parked as a dependency";
    final List<Modification> parkedMods = generatemods("description", parkedDescription);
    final List<Modification> unwoundMods =
        generatemods("description", "the replay of this change is unwound once it is applied");

    /*
     * The change whose replay fails is the barrier the parked change waits behind, so its
     * budget must not be spent while this test is setting up: it is shortened once the
     * change which was parked is back, and put back in the finally below.
     */
    setReplayGiveUpDelay("unlimited");
    try
    {
      /*
       * One replay thread, so that the change which is parked and the replay which is
       * unwound after it are the same thread's: a parked change is left owned by the thread
       * which parked it, and this is about a thread which does not come back to it.
       */
      setNumUpdateReplayThreads(1);
      try
      {
        /*
         * The change which is parked is published to the replication server rather than
         * handed to the domain: a change which travelled a session is one the replication
         * server sends again over the next session of this domain, and over nothing else -
         * which is what the restart the give-back asks for has to be pinned against.
         */
        final ReplicationBroker broker =
            openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
        try
        {
          /*
           * A change whose replay failed stays listed and uncommitted - it is what holds this
           * domain's ServerState back - and stays among the changes the newer ones are checked
           * against, so a change which follows it on the same entry has to wait for it. It has
           * to be a change which is asked for again rather than stepped over: one whose
           * operation is built and then refused, since #928 has a modify whose entry DN does
           * not parse reported once and recorded as replayed.
           */
          deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
              () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
              "the change whose replay fails must be listed as one which is not in the data");

          // The change which is parked as waiting for it by the replay thread it was given to.
          broker.publish(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
          assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
              "a change which waits for one that is not in the data must be parked");

          /*
           * How many restarts in a row the session has been through, the barrier's among
           * them: the restart the give-back asks for on the road out of a JVM which has run
           * out of memory is owed no backoff, and a restart which waits the backoff out is
           * the one road which moves this count. It can not move otherwise between here and
           * the reading below: a replay which made it puts the count back to zero only once
           * nothing is failing anymore, and the barrier keeps failing until it is given up.
           */
          final int restarts = domain.getConsecutiveSessionRestarts();

          /*
           * The replay which is unwound while that same thread still holds the parked change.
           * It applies its change, and the ack of its delivery is where the JVM runs out of
           * memory: that is the one throw from the ack which is not caught, past every catch
           * the replay itself has, and it leaves replay() with the change this thread was
           * replaying committed and owned by nobody. What the give-back on the way out finds
           * to give back is the parked change alone, and the restart it asks for is the one
           * road which has that change delivered again: the road a failed replay takes to
           * ask for its own change is not run. An Error met replaying a change does not get
           * here, and neither does any other throw from the ack: both are reported and the
           * replay carries on to the road the change itself decided (issue #922).
           *
           * It is made on another entry, so that it is replayed rather than parked in its
           * turn, and it is delivered until the parked change has been given back: a delivery
           * of a change which is being replayed, or was applied, is refused as the duplicate
           * it is.
           */
          deliverUntilMonitorReaches(domain, "dependent-changes-size", 0,
              () -> new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, other, unwoundMods, otherUUID),
              "a change parked by a replay which was unwound must be given back");

          /*
           * The thread which met the error is gone - an OutOfMemoryError ends the replay
           * thread which met it (issue #923) - and it was the whole pool, so the delivery the
           * restarted session brings waits in the replay queue: the pool is brought back for
           * it, at any number now that the thread it had to be the same as is over. The change
           * is parked again by the thread which takes it, the barrier being still there: that
           * it is parked at all is what says a new session delivered it.
           */
          setNumUpdateReplayThreads(2);
          assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
              "the change which was given back must be delivered again by the session which"
                  + " was restarted for it");
          assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
              "a thread an OutOfMemoryError is ending must not wait the backoff out before"
                  + " it restarts the session for the changes it gave back");

          /*
           * The barrier is lifted, which lets the ServerState past it and hands the parked
           * change to the thread which cleared it.
           */
          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
          checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
              "the change which was parked must be applied by the delivery which took it over");
        }
        finally
        {
          broker.stop();
        }
        assertEquals(resetNumUpdateReplayThreads(), 0,
            "the number of replay threads could not be put back");
      }
      finally
      {
        /*
         * Best-effort on the way out of a red, with no assertion to replace the failure
         * which is being reported: on the normal road the number is put back, and checked,
         * above.
         */
        resetNumUpdateReplayThreads();
      }
    }
    finally
    {
      try
      {
        /*
         * A red before the barrier was given up would leave it listed for the rest of this
         * class: it never travelled the replication server, so no session restart brings a
         * delivery which could give it up, and a change which is failing keeps every later
         * session restart at its backoff and the ServerState behind it. The failure which
         * is being reported is the one that matters, so this does not replace it.
         */
        if (!domain.getServerState().cover(failing))
        {
          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
        }
      }
      catch (Throwable cleanupFailure)
      {
        logger.error(LocalizableMessage.raw(
            "the barrier of aChangeParkedByAnUnwoundReplayIsDeliveredAgain could not be given"
                + " up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
      }
      finally
      {
        resetReplayGiveUpDelay();
      }
    }
  }

  /**
   * Gives up on the change no delivery can replay: the budget is shortened to nothing, so
   * that the change is given up on as soon as one more delivery of it fails. Nothing sends
   * that change again - it never travelled a session - so its deliveries are made here.
   */
  private void giveUpOn(final LDAPReplicationDomain domain, final CSN failing, final DN dn,
      final List<Modification> mods, final String entryUUID) throws Exception
  {
    setReplayGiveUpDelay("0ms");
    TestTimer timer = new TestTimer.Builder()
      .maxSleep(120, SECONDS)
      .sleepTimes(500, MILLISECONDS)
      .toTimer();
    timer.repeatUntilSuccess(new CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        if (!domain.getServerState().cover(failing))
        {
          domain.processUpdate(new ModifyMsgWhoseOperationRefusesAControl(failing, dn, mods, entryUUID));
        }
        assertTrue(domain.getServerState().cover(failing),
            "the change no delivery can replay must be given up on");
      }
    });
  }

  /**
   * Test case for [Issue 986]: a change parked as waiting for another one is given back
   * when the replay thread which parked it is stopped with the pool.
   * <p>
   * A parked change stays owned by the replay thread which parked it while that thread
   * goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
   * what hands it out again, to whichever replay thread clears the change it was waiting
   * for. Changing the number of replay threads stops the whole pool and creates another
   * one, so a thread which parked a change and went back to the queue is joined while it
   * is idle, and it would end still recorded as the owner of that change - a thread which
   * does not exist anymore, while every redelivery of a change a replay thread owns is
   * refused as a duplicate. On a domain which then goes quiet that change is where this
   * replica's ServerState, and every change behind it from every master, stops.
   */
  @Test
  public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception
  {
    testSetUp("aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain"));

    final DN waitedOn = addEntryForChange("user.986.1");
    final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
    final DN other = addEntryForChange("user.986.2");
    final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
    domain.resetUnreplayedChangeAlertThrottle();
    final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "no change of this domain is waiting for another one when this test starts");

    final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
    final CSN failing = gen.newCSN();
    final CSN parked = gen.newCSN();
    final CSN applied = gen.newCSN();
    final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
    final String parkedDescription = "the change which was parked by a thread the pool stopped";
    final List<Modification> parkedMods = generatemods("description", parkedDescription);
    final String appliedDescription = "the change which is applied before the pool is stopped";
    final List<Modification> appliedMods = generatemods("description", appliedDescription);

    /*
     * One replay thread, so that the change which is parked is parked by the thread the
     * pool then stops: a parked change is left owned by the thread which parked it, and
     * this is about a thread which is not there anymore to be given it back.
     */
    setNumUpdateReplayThreads(1);
    try
    {
      /*
       * A change whose replay failed stays listed and uncommitted - it is what holds this
       * domain's ServerState back - and stays among the changes the newer ones are checked
       * against, so a change which follows it on the same entry has to wait for it.
       */
      deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
          () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
          "the change whose replay fails must stay listed as one which is not in the data");

      /*
       * The change which is parked as waiting for it. The replay thread it was given to
       * parks it and goes back to the pool: nothing is waiting for it there, so it is idle
       * and it still owns the change it parked.
       */
      deliverUntilMonitorReaches(domain, "dependent-changes-size", 1,
          () -> new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID),
          "a change which waits for one that is not in the data must be parked");

      /*
       * A change which is applied, on another entry so that it is replayed rather than
       * parked in its turn. It is there for the count of the deliveries this session took
       * off it, replayed-updates: a session which is started counts from zero, so that
       * count going back to zero below is what says the session was restarted - and it is
       * not left to the give-back alone to put the count above zero before the pool is
       * stopped. The restart the change whose replay failed asked for is over by now: the
       * thread which ran it is the one which parked the change above.
       */
      domain.processUpdate(new ModifyMsg(applied, other, appliedMods, otherUUID));
      checkEntryHasAttributeValue(other, "description", appliedDescription, 30,
          "the change made on the other entry must be applied");
      // Counted once its ack is out, which is after the change is in the data.
      final TestTimer counted = new TestTimer.Builder()
        .maxSleep(30, SECONDS)
        .sleepTimes(200, MILLISECONDS)
        .toTimer();
      counted.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          assertTrue(getMonitorAttrValue(baseDN, "replayed-updates") > 0,
              "the session must have counted the deliveries it took off before the pool is stopped");
        }
      });

      /*
       * How many restarts in a row the session has been through, the barrier's among them:
       * the restart the give-back of a stopped thread asks for is owed no backoff - what
       * went away is a replay thread, not the backend - and a restart which waits the
       * backoff out is the one road which moves this count. It can not move otherwise
       * between here and the reading below: a replay which made it puts the count back to
       * zero only once nothing is failing anymore, and the barrier keeps failing until it
       * is given up, below. The restart the barrier asked for is over: the thread which
       * ran it is the one which parked the change and applied the other, above.
       */
      final int restarts = domain.getConsecutiveSessionRestarts();

      /*
       * The pool is stopped and created again, the way an administrator changing the
       * number of replay threads has it: the thread which parked the change is joined
       * where it waits for the next delivery, and the change it owns is nobody's.
       */
      setNumUpdateReplayThreads(2);

      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 0,
          "a change parked by a replay thread the pool stopped must be given back");
      /*
       * The give-back alone leaves the count where it was, plus the change it handed back:
       * only a session which is started counts from zero, and the changes of this test
       * never travelled a session, so nothing is delivered over the one which is started
       * before the deliveries made below.
       */
      assertMonitorAttrValueEventually(baseDN, "replayed-updates", 0,
          "the session must be restarted for the changes which were given back");
      assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
          "the restart run for the changes a stopped thread gave back must not wait the"
              + " backoff out: what went away is a replay thread, not the backend");

      /*
       * Nothing sends these changes again - they never travelled a session - so the
       * deliveries which take over from the ones the stopped pool left behind are made
       * here. The change no delivery can replay is given up on, which lets the ServerState
       * past it, and the change which was parked behind it is applied.
       */
      setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
      TestTimer timer = new TestTimer.Builder()
        .maxSleep(120, SECONDS)
        .sleepTimes(500, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          final ServerState state = domain.getServerState();
          if (!state.cover(failing))
          {
            domain.processUpdate(
                new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID));
          }
          if (!state.cover(parked))
          {
            domain.processUpdate(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
          }
          assertTrue(state.cover(parked),
              "the change which was given back must be replayed by the delivery which takes it over");
        }
      });
      checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
          "the change which was parked must be applied by the delivery which took it over");
      assertEquals(resetNumUpdateReplayThreads(), 0,
          "the number of replay threads could not be put back");
    }
    finally
    {
      try
      {
        /*
         * A red before the barrier was given up would leave it listed for the rest of this
         * class: it never travelled the replication server, so no session restart brings a
         * delivery which could give it up, and a change which is failing keeps every later
         * session restart at its backoff and the ServerState behind it. The failure which
         * is being reported is the one that matters, so this does not replace it.
         */
        if (!domain.getServerState().cover(failing))
        {
          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
        }
      }
      catch (Throwable cleanupFailure)
      {
        logger.error(LocalizableMessage.raw(
            "the barrier of aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain could not be"
                + " given up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
      }
      finally
      {
        try
        {
          resetReplayGiveUpDelay();
        }
        finally
        {
          /*
           * Best-effort on the way out of a red, with no assertion to replace the failure
           * which is being reported: on the normal road the number is put back, and checked,
           * above.
           */
          resetNumUpdateReplayThreads();
        }
      }
    }
  }

  /**
   * Delivers a change until a monitor attribute of the domain reaches the expected value.
   * <p>
   * A delivery is dropped rather than queued while the listener thread is down, which it
   * is for as long as a recovery is restarting the session, so a change which has to reach
   * a replay thread is delivered until it does. A delivery of a change a replay thread
   * owns is refused as the duplicate it is, so the deliveries which follow the one that
   * was taken cost nothing.
   */
  private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain,
      final String attributeName, final long expected,
      final Supplier<? extends LDAPUpdateMsg> delivery, final String message) throws Exception
  {
    TestTimer timer = new TestTimer.Builder()
      .maxSleep(20, SECONDS)
      .sleepTimes(500, MILLISECONDS)
      .toTimer();
    timer.repeatUntilSuccess(new CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        if (getMonitorAttrValue(baseDN, attributeName) != expected)
        {
          domain.processUpdate(delivery.get());
        }
        assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message);
      }
    });
  }

  /**
   * Sets how many replay threads this server runs, the way an administrator would: the
   * pool is stopped and created again with that number.
   */
  private static void setNumUpdateReplayThreads(int threads) throws Exception
  {
    assertEquals(TestCaseUtils.applyModifications(true,
        "dn: " + SYNCHRO_PLUGIN_DN,
        "changetype: modify",
        "replace: ds-cfg-num-update-replay-threads",
        "ds-cfg-num-update-replay-threads: " + threads), 0,
        "the number of replay threads could not be changed");
  }

  /**
   * Puts the number of replay threads back to what this server computes for itself, which
   * is what it runs with when the configuration carries no number of its own.
   *
   * @return the result code of the change, so that a finally can call this without an
   *         assertion which would replace the failure it is on the way out of
   */
  private static int resetNumUpdateReplayThreads() throws Exception
  {
    return TestCaseUtils.applyModifications(true,
        "dn: " + SYNCHRO_PLUGIN_DN,
        "changetype: modify",
        "delete: ds-cfg-num-update-replay-threads");
  }

  /** Adds the entry a change of these tests is made on. */
  private DN addEntryForChange(String uid) throws Exception
  {
    return TestCaseUtils.addEntry(
        "dn: uid=" + uid + "," + baseDN,
        "objectClass: top",
        "objectClass: person",
        "objectClass: organizationalPerson",
        "objectClass: inetOrgPerson",
        "uid: " + uid,
        "cn: Aaccf Amar",
        "sn: Amar").getName();
  }

  /** A delivery of a change whose replay does not run to its end. */
  private interface FailingDelivery
  {
@@ -3603,16 +4106,7 @@
  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();
    final DN dn = addEntryForChange(uid);
    final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();

    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
@@ -3842,6 +4336,36 @@
  }

  /**
   * A ModifyMsg whose ack runs out of memory on the way out of a replay which applied it.
   * <p>
   * The change is committed before the ack of its delivery is published, and the ack is
   * where the JVM runs out of memory: the one throw from there which is not caught, so the
   * replay is unwound with the change it was replaying in the data and owned by nobody -
   * commit() cleared the owner - and the thread ends on the error. What the give-back on
   * the way out of {@code replay()} has left to give back is the changes this thread parked
   * as waiting for another one, and the restart it asks for them is the one which is run:
   * the road a failed replay takes to ask for its own change again is not on the way.
   * <p>
   * Nothing on the way in reads what throws here: a message handed to the domain rather
   * than published is not one this server acknowledges to anybody.
   */
  private static final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
  {
    private ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
        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 applied 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

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

New file
@@ -0,0 +1,184 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static org.assertj.core.api.Assertions.*;
import static org.opends.server.TestCaseUtils.*;

import java.util.ArrayList;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.backends.MemoryBackend;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 * Tests that {@link MultimasterReplication#forEachDomain} gives every domain its turn
 * whatever one of them threw, and reports the failures the way it says it does: the first
 * is thrown once the loop is over, the others are suppressed under it where it records
 * suppression, and an instance thrown by two domains - the error a JVM out of memory hands
 * out as often as it is asked for one - is thrown once and suppresses nothing (issue #986).
 * <p>
 * Two domains, on backends of their own, neither of them started: the action is what
 * throws, and it never touches the domain it is given.
 */
@SuppressWarnings("javadoc")
public class ForEachDomainTest extends ReplicationTestCase
{
  private static final int RS_ID = 613;
  private static final String SECOND_BACKEND_ID = "test2";
  private static final String SECOND_ROOT_DN_STRING = "o=" + SECOND_BACKEND_ID;

  private DN firstBaseDN;
  private DN secondBaseDN;
  private ReplicationServer replicationServer;
  private LDAPReplicationDomain firstDomain;
  private LDAPReplicationDomain secondDomain;

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
    firstBaseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    secondBaseDN = DN.valueOf(SECOND_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    TestCaseUtils.initializeMemoryBackend(SECOND_BACKEND_ID, SECOND_ROOT_DN_STRING, true);

    final int rsPort = TestCaseUtils.findFreePort();
    replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
        rsPort, "forEachDomainTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
    final SortedSet<String> replServers = new TreeSet<>();
    replServers.add("localhost:" + rsPort);
    firstDomain = MultimasterReplication.createNewDomain(new DomainFakeCfg(firstBaseDN, 1, replServers));
    secondDomain = MultimasterReplication.createNewDomain(new DomainFakeCfg(secondBaseDN, 2, replServers));
  }

  @AfterMethod
  public void tearDown() throws Exception
  {
    try
    {
      MultimasterReplication.deleteDomain(firstBaseDN);
      MultimasterReplication.deleteDomain(secondBaseDN);
    }
    finally
    {
      try
      {
        remove(replicationServer);
      }
      finally
      {
        final MemoryBackend backend = (MemoryBackend) getServerContext().getBackendConfigManager()
            .getLocalBackendById(SECOND_BACKEND_ID);
        if (backend != null)
        {
          backend.clearMemoryBackend();
          backend.finalizeBackend();
          getServerContext().getBackendConfigManager().deregisterLocalBackend(backend);
        }
      }
    }
  }

  @Test
  public void theFirstFailureIsThrownOnceTheLoopIsOverWithTheOthersSuppressedUnderIt()
  {
    final RuntimeException first = new RuntimeException("first");
    final RuntimeException second = new RuntimeException("second");
    final List<LDAPReplicationDomain> visited = new ArrayList<>();

    final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
    {
      visited.add(domain);
      throw visited.size() == 1 ? first : second;
    }));

    assertThat(thrown).as("the failure thrown once the loop is over must be the first one met")
        .isSameAs(first);
    assertThat(thrown.getSuppressed()).as("the failure met after the first must be suppressed under it")
        .containsExactly(second);
    assertThat(visited).as("every domain must get its turn whatever the one before it threw")
        .containsExactlyInAnyOrder(firstDomain, secondDomain);
  }

  @Test
  public void aFailureThrownByTwoDomainsIsThrownOnceAndSuppressesNothing()
  {
    // The error a JVM out of memory prepared beforehand is handed out to every domain alike.
    final OutOfMemoryError theOneError = new OutOfMemoryError("prepared beforehand");
    final List<LDAPReplicationDomain> visited = new ArrayList<>();

    final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
    {
      visited.add(domain);
      throw theOneError;
    }));

    assertThat(thrown).as("the failure thrown once the loop is over must be the one both domains threw")
        .isSameAs(theOneError);
    assertThat(thrown.getSuppressed()).as("a failure must not be suppressed under itself").isEmpty();
    assertThat(visited).as("every domain must get its turn whatever the one before it threw")
        .containsExactlyInAnyOrder(firstDomain, secondDomain);
  }

  @Test
  public void aFirstFailureWhichRecordsNoSuppressionStillGivesEveryDomainItsTurn()
  {
    final Error first = new ErrorWhichRecordsNoSuppression();
    final RuntimeException second = new RuntimeException("second");
    final List<LDAPReplicationDomain> visited = new ArrayList<>();

    final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
    {
      visited.add(domain);
      if (visited.size() == 1)
      {
        throw first;
      }
      throw second;
    }));

    assertThat(thrown).as("the failure thrown once the loop is over must be the first one met")
        .isSameAs(first);
    assertThat(thrown.getSuppressed()).as("a failure which records no suppression must have none recorded")
        .isEmpty();
    assertThat(visited).as("every domain must get its turn whatever the one before it threw")
        .containsExactlyInAnyOrder(firstDomain, secondDomain);
  }

  /**
   * An error which keeps no record of the failures suppressed under it: the shape of the
   * error a JVM out of memory prepared beforehand, which was made without its constructor
   * and so keeps no list to record them in.
   */
  private static final class ErrorWhichRecordsNoSuppression extends Error
  {
    private static final long serialVersionUID = 1L;

    private ErrorWhichRecordsNoSuppression()
    {
      super("records no suppression", null, false, false);
    }
  }
}

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

New file
@@ -0,0 +1,82 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.types.Modification;

/**
 * A ModifyMsg whose ack runs out of memory on the way out of a replay which applied it.
 * <p>
 * The change is committed before the ack of its delivery is published, and the ack is
 * where the JVM runs out of memory: the one throw from there which is not caught, so the
 * replay is unwound with the change it was replaying in the data and owned by nobody -
 * {@code commit()} cleared the owner - and the thread ends on the error. What the give-back
 * on the way out of {@code replay()} has left to give back is the changes this thread
 * parked as waiting for another one, and the restart it asks for them is the one which is
 * run: the road a failed replay takes to ask for its own change again is not on the way.
 * <p>
 * Given a flag, the ack sets it before it throws. Handed the flag the replay reads as "this
 * thread is stopping", that stops the thread while its change is being acknowledged: applied
 * and committed under a running thread, unwound under a stopping one, the way a thread of
 * the pool is stopped when their number is changed while its ack is on its way out. A flag
 * set before the replay would not reach here with the change applied: the replay abandons
 * the change unapplied at the top of its first attempt, still owned by this thread, and the
 * abandon road asks for a restart of its own next to the one the give-back asks for.
 * <p>
 * Nothing on the way in reads what throws here: a message handed to the domain rather than
 * published is not one this server acknowledges to anybody. The twin of the fixture
 * {@code UpdateOperationTest} unwinds a replay thread with.
 */
final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
{
  /** The flag the ack sets before it throws, or {@code null} for none. */
  private final AtomicBoolean setByTheAck;

  ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
      CSN csn, DN dn, List<Modification> mods, String entryUUID)
  {
    this(csn, dn, mods, entryUUID, null);
  }

  /**
   * @param setByTheAck the flag the ack sets before it throws: the one the replay reads as
   *          "this thread is stopping", to stop the thread while its applied change is
   *          being acknowledged
   */
  ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
      CSN csn, DN dn, List<Modification> mods, String entryUUID, AtomicBoolean setByTheAck)
  {
    super(csn, dn, mods, entryUUID);
    this.setByTheAck = setByTheAck;
  }

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

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

New file
@@ -0,0 +1,67 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static org.opends.server.protocols.internal.InternalClientConnection.*;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.core.ModifyOperation;
import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.protocol.ModifyContext;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
import org.opends.server.types.Control;
import org.opends.server.types.Modification;
import org.opends.server.types.RawModification;

/**
 * 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 on -
 * and the list of request controls it carries can not be added to, so the ManageDsaIT
 * control the replay puts on every operation throws before
 * {@code OperationContext.getCSN(op)} is reached: the change fails, stays listed and
 * uncommitted, and is asked for again. Such a message can not travel the protocol:
 * {@code ModifyMsg.createOperation()} builds an operation whose controls can be added to,
 * so this one is handed to the domain rather than published. The twin of the fixture
 * {@code UpdateOperationTest} holds its barrier with.
 */
final class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
{
  ModifyMsgWhoseOperationRefusesAControl(
      CSN csn, DN dn, List<Modification> mods, String entryUUID)
  {
    super(csn, dn, mods, entryUUID);
  }

  @Override
  public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
  {
    final ModifyOperation op = new ModifyOperationBasis(connection, nextOperationID(),
        nextMessageID(), Collections.<Control>emptyList(),
        ByteString.valueOfUtf8(getDN().toString()), new ArrayList<RawModification>());
    op.setAttachment(OperationContext.SYNCHROCONTEXT,
        new ModifyContext(getCSN(), getEntryUUID()));
    return op;
  }
}

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

New file
@@ -0,0 +1,373 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static org.assertj.core.api.Assertions.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.testng.Assert.*;

import java.util.ArrayList;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.types.Entry;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 * Tests who runs the session restart asked for the changes a replay which is unwound had
 * parked as waiting for another change (issue #954).
 * <p>
 * The thread which gave them back runs it itself, so that the delivery they wait for is
 * not left to the next tick of the state checkpointer - unless that thread is stopping, in
 * which case it leaves the request standing the way a stopping thread leaves the request it
 * makes for the change it abandons: the threads of the pool are stopped one after the
 * other and joined, and each running a restart on its way out would have the configuration
 * change which is stopping them wait for one restart per thread.
 * <p>
 * The replay runs on the thread of the test, which is what says whether that thread is
 * stopping, and the error which unwinds it is caught here rather than ending a replay
 * thread. The thread which is stopping is stopped by the ack of the change it applied, and
 * the case asserts that the change was applied: stopped before the replay, it abandons the
 * change unapplied at the top of its first attempt, and the abandon road asks for a restart
 * of its own next to the give-back's, which would then be pinned by nothing - that road is
 * the third case's, which pins the abandon arm of the catch. The restart is asked to fail
 * once, so that the thread which ran it is the one which met the failure: a replay thread
 * carries the failure out with the error it is ending on, the state checkpointer reports
 * it. The restart is asked for at once, without the backoff, and the count of the restarts
 * in a row says so: only the restart run again after the failure waits its backoff out.
 */
@SuppressWarnings("javadoc")
public class ParkedChangeGiveBackTest extends ReplicationTestCase
{
  private static final int RS_ID = 612;
  private static final int DS_ID = 1;
  private static final AtomicBoolean RUNNING = new AtomicBoolean(false);

  /**
   * How long the state checkpointer is given to run a request left standing and report
   * the failure it was asked to meet: its next tick, at most a second away, and the
   * report.
   */
  private static final long CHECKPOINTER_BOUND_IN_MS = 5000;
  /**
   * How long the session is given to come back once a restart failed: the backoff a
   * restart which follows a failed one is owed - the second in a row here, two seconds -
   * and the tick of the state checkpointer which runs it.
   */
  private static final long RESTART_BOUND_IN_MS = 10000;

  private DN baseDN;
  private ReplicationServer replicationServer;
  private LDAPReplicationDomain domain;
  private TestSynchronousReplayQueue queue;
  private CSNGenerator gen;

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
    baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);

    final int rsPort = TestCaseUtils.findFreePort();
    replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
        rsPort, "parkedChangeGiveBackTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));

    final SortedSet<String> replServers = new TreeSet<>();
    replServers.add("localhost:" + rsPort);
    final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID, replServers);
    conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);
    queue = new TestSynchronousReplayQueue();
    domain = MultimasterReplication.createNewDomain(conf, queue);
    domain.start();
    assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
    gen = new CSNGenerator(201, 0);
  }

  @AfterMethod
  public void tearDown() throws Exception
  {
    try
    {
      domain.failNextSessionRestarts(0);
      MultimasterReplication.deleteDomain(baseDN);
    }
    finally
    {
      remove(replicationServer);
    }
  }

  /**
   * The thread which gave a parked change back runs the restart it asked for, and meets
   * the failure that restart was asked to meet: the failure comes out with the error which
   * unwound the replay, as one it suppressed, and the restart which is run again after it
   * brings the session back before the replay returns. The restart it asked for is run at
   * once: the one run again after the failure is the one which waits its backoff out, and
   * that is the one move of the count of the restarts in a row.
   */
  @Test(timeOut = 120_000)
  public void theThreadWhichGaveBackAParkedChangeRunsTheRestartItAskedFor() throws Exception
  {
    parkAChangeBehindABarrier(addEntry("waitedOn"));
    final Entry other = addEntry("other");
    final int restartsBefore = domain.getConsecutiveSessionRestarts();
    domain.failNextSessionRestarts(1);

    final OutOfMemoryError unwinding = unwindAReplay(other, Stopped.NEVER);

    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "the change parked by the replay which was unwound must be given back");
    assertThat(injectedRestartFailuresAmong(unwinding.getSuppressed()))
        .as("the thread which gave the parked change back must have run the restart it"
            + " asked for, and met the failure that restart was asked to meet")
        .hasSize(1);
    assertEquals(domain.getSessionRestartFailuresLeft(), 0,
        "the restart which was asked to fail never ran");
    awaitConnected(RESTART_BOUND_IN_MS, "the restart run again after the one which failed did"
        + " not bring the session back");
    assertEquals(domain.getConsecutiveSessionRestarts(), restartsBefore + 1,
        "the restart the give-back asked for must be run at once, without the backoff: the one"
            + " run again after the failure is the one which waits it out");
  }

  /**
   * A thread which is stopping leaves the restart it asked for standing, and the state
   * checkpointer runs it: the failure that restart was asked to meet is the checkpointer's
   * to report, and nothing of it comes out with the error which unwound the replay.
   * <p>
   * The thread is stopped by the ack of the change it applied, so the change is committed
   * and owned by nobody by the time the replay is unwound, and the request the give-back
   * makes for the parked change is the one request which stands for the checkpointer. The
   * case asserts that road: a thread stopped before the replay hands its change back on a
   * road of its own, which asks for a restart next to the give-back's and would stand in
   * for it.
   */
  @Test(timeOut = 120_000)
  public void aStoppingThreadLeavesTheRestartItAskedForToTheStateCheckpointer() throws Exception
  {
    parkAChangeBehindABarrier(addEntry("waitedOn"));
    final Entry other = addEntry("other");
    final int reportedBefore = restartFailureReports().size();
    final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
    final int restartsBefore = domain.getConsecutiveSessionRestarts();
    domain.failNextSessionRestarts(1);

    final OutOfMemoryError unwinding = unwindAReplay(other, Stopped.BY_THE_ACK);

    assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore + 1,
        "the change must be applied and its replay unwound by the ack, not abandoned unapplied");
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "the change parked by the replay which was unwound must be given back");
    assertThat(injectedRestartFailuresAmong(unwinding.getSuppressed()))
        .as("a thread which is stopping must not run the restart it asked for")
        .isEmpty();

    final long deadline = System.currentTimeMillis() + CHECKPOINTER_BOUND_IN_MS;
    while (restartFailureReports().size() == reportedBefore)
    {
      assertTrue(System.currentTimeMillis() < deadline, "the state checkpointer did not run"
          + " the restart the stopping thread left standing: the failure that restart was"
          + " asked to meet was never reported");
      Thread.sleep(50);
    }
    awaitConnected(RESTART_BOUND_IN_MS,
        "the session was not brought back after the restart which failed");
    assertEquals(domain.getConsecutiveSessionRestarts(), restartsBefore + 1,
        "the restart the give-back asked for must be run at once, without the backoff: the one"
            + " run again after the failure is the one which waits it out");
  }

  /**
   * A thread stopped before its first attempt hands back the change it did not apply, the
   * way a stopping thread abandons a replay: the ack of an abandoned change is published
   * all the same, and it is what runs out of memory here, so the abandon road taken is the
   * one of the catch - the change is still this thread's when the replay is unwound.
   */
  @Test(timeOut = 120_000)
  public void aStoppingThreadHandsBackTheChangeItDidNotApply() throws Exception
  {
    parkAChangeBehindABarrier(addEntry("waitedOn"));
    final Entry other = addEntry("other");
    final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
    final CSN abandoned = gen.newCSN();

    unwindAReplay(other, abandoned, Stopped.BEFORE_THE_REPLAY);

    assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore,
        "a thread stopped before the replay must not apply the change");
    assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), abandoned))
        .as("a stopping thread must hand back the change it did not apply")
        .isNotEmpty();
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "the change parked by the thread which is stopping must be given back");
  }

  private void awaitConnected(long boundInMs, String message) throws Exception
  {
    final long deadline = System.currentTimeMillis() + boundInMs;
    while (!domain.isConnected())
    {
      assertTrue(System.currentTimeMillis() < deadline, message);
      Thread.sleep(50);
    }
  }

  private Entry addEntry(String cn) throws Exception
  {
    return TestCaseUtils.addEntry(
        "dn: cn=" + cn + "," + TEST_ROOT_DN_STRING,
        "objectClass: top",
        "objectClass: person",
        "cn: " + cn,
        "sn: " + cn);
  }

  /**
   * Replays, on the thread of this test, a change whose replay fails and stays listed - its
   * operation is built and then refused, so it is asked for again rather than stepped over -
   * and then a change on the same entry, which is parked as waiting for it and owned by this
   * thread from then on. The restart the failed change asks for is run on this thread as
   * well, so the session is back once this returns.
   */
  private void parkAChangeBehindABarrier(Entry entry) throws Exception
  {
    final String entryUUID = getEntryUUID(entry.getName());
    final CSN failing = gen.newCSN();
    replayMsg(new ModifyMsgWhoseOperationRefusesAControl(failing, entry.getName(),
        generatemods("description", "the replay of this change fails"), entryUUID), RUNNING);
    assertFalse(domain.getServerState().cover(failing),
        "the change whose replay fails must stay listed as one which is not in the data");
    awaitConnected(RESTART_BOUND_IN_MS,
        "the session was not brought back for the change whose replay failed");

    replayMsg(new ModifyMsg(gen.newCSN(), entry.getName(),
        generatemods("description", "the change which was parked as a dependency"), entryUUID),
        RUNNING);
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
        "a change which waits for one that is not in the data must be parked");
  }

  /** When the thread a replay runs on is stopped, if it is. */
  private enum Stopped
  {
    /** It is running throughout. */
    NEVER,
    /** By the ack of the change it applied: the change is committed, and the replay is unwound. */
    BY_THE_ACK,
    /**
     * Before the replay: the change is abandoned unapplied at the top of its first attempt,
     * and the ack which says so is what runs out of memory.
     */
    BEFORE_THE_REPLAY
  }

  private OutOfMemoryError unwindAReplay(Entry entry, Stopped stopped) throws Exception
  {
    return unwindAReplay(entry, gen.newCSN(), stopped);
  }

  /**
   * Replays, on the thread of this test, a change whose ack runs out of memory, and returns
   * the error the replay was unwound on.
   *
   * @param stopped when the thread the replay runs on is stopped: running until the ack,
   *          the change is applied rather than abandoned at the top of its first attempt
   */
  private OutOfMemoryError unwindAReplay(Entry entry, CSN csn, Stopped stopped) throws Exception
  {
    // Fresh for every replay: the ack is what sets it, and nothing puts it back.
    final AtomicBoolean stopping = new AtomicBoolean(stopped == Stopped.BEFORE_THE_REPLAY);
    try
    {
      replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(csn, entry.getName(),
          generatemods("description", "the replay of this change is unwound by its ack"),
          getEntryUUID(entry.getName()), stopped == Stopped.BY_THE_ACK ? stopping : null), stopping);
    }
    catch (OutOfMemoryError unwinding)
    {
      // The error is the fixture's own, and this is the thread it would have ended.
      return unwinding;
    }
    throw new AssertionError("the replay was not unwound: the ack of the delivery must run out of memory");
  }

  /** The throwables among the provided ones which a restart threw because a test asked it to. */
  private static List<Throwable> injectedRestartFailuresAmong(Throwable[] suppressed)
  {
    final List<Throwable> injected = new ArrayList<>();
    for (Throwable t : suppressed)
    {
      if (t instanceof IllegalStateException && t.getMessage().contains("as a test asked"))
      {
        injected.add(t);
      }
    }
    return injected;
  }

  /** The records of the error log which report a session restart of this domain that threw. */
  private List<String> restartFailureReports()
  {
    final List<String> records = new ArrayList<>();
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    {
      if (record.contains("msgID=" + ERR_REPLAY_SESSION_RESTART_FAILED.ordinal())
          && record.contains(baseDN.toString()))
      {
        records.add(record);
      }
    }
    return records;
  }

  /** The records of the error log which carry the provided message for the provided change. */
  private static List<String> errorLogRecordsOf(int msgId, CSN csn)
  {
    final List<String> records = new ArrayList<>();
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    {
      if (record.contains("msgID=" + msgId) && record.contains(csn.toString()))
      {
        records.add(record);
      }
    }
    return records;
  }

  private void replayMsg(UpdateMsg updateMsg, AtomicBoolean stopping) throws InterruptedException
  {
    domain.processUpdate(updateMsg);
    final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage();
    domain.markInProgress(ldapUpdate);
    domain.replay(ldapUpdate, stopping);
  }
}

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

@@ -15,6 +15,7 @@
 */
package org.opends.server.replication.plugin;

import static java.util.Collections.*;
import static org.testng.Assert.*;

import java.util.NoSuchElementException;
@@ -670,10 +671,12 @@

  /**
   * 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).
   * replaying: it is handed to whichever thread clears what it waits for, so the give-back
   * of the change a replay was unwound on 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) - which is why the parked ones are given back on a road of their
   * own, {@link RemotePendingChanges#releaseParkedChangesOwnedByCurrentThread()}, where
   * both happen in one step (issue #954).
   * <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
@@ -799,6 +802,145 @@
        "a change which has been handed out must not be handed out again");
  }

  /**
   * Test case for [Issue 954]: a change parked as waiting for another one is given back
   * when the replay which parked it is unwound.
   * <p>
   * A parked change stays owned by the thread which parked it, and is handed out again by
   * {@link RemotePendingChanges#getNextUpdate()} to whichever thread clears the change it
   * waits for. A replay which is unwound - a JVM out of memory, a throw from what the replay
   * runs once the ack of its delivery is out - leaves that thread without a road back to the
   * change: it takes the next delivery off the queue instead. Nothing else asks for the
   * change either, since every redelivery of a change a replay thread owns is refused as a
   * duplicate, so this domain's ServerState would stay behind it until some other change is
   * replayed on this domain.
   */
  @Test
  public void aChangeParkedByAReplayWhichIsUnwoundIsGivenBack() 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 replay which parked it is unwound, so it gives back what it still owns.
    assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(),
        singletonList(renamed), "the change this thread parked must be given back");

    assertEquals(pendingChanges.getDependentChangesSize(), 0,
        "a change which was given back must not be left waiting for a thread to hand it out");
    assertEquals(pendingChanges.getQueueSize(), 2, "the change must stay listed as pending");
    assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed");
    assertEquals(pendingChanges.changesInProgressSize(), 2,
        "a change which is not in the data yet must stay a dependency of the changes which follow it");

    assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
        "the change the unwound replay gave back must be taken over by the next delivery");
  }

  /**
   * Test case for [Issue 954]: a change which was given back is handed out by the next
   * delivery of it and by nothing else.
   * <p>
   * Unparking it and giving it back is one step under both locks for that reason: a
   * change which was released while it is still listed as waiting would be replayed by
   * the thread {@link RemotePendingChanges#getNextUpdate()} hands it to and by the thread
   * which takes over the delivery which follows - the double replay the ownership is
   * there to prevent (OPENDJ-1115).
   */
  @Test
  public void aParkedChangeWhichWasGivenBackIsNotHandedOutAsADependency() 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));

    final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
    assertTrue(pendingChanges.putRemoteUpdate(rename));
    assertTrue(pendingChanges.markInProgress(rename));
    assertTrue(pendingChanges.checkDependencies(rename));
    pendingChanges.releaseParkedChangesOwnedByCurrentThread();

    // The change it was waiting for is replayed, which is what used to hand it out.
    pendingChanges.commit(deleted);

    assertNull(pendingChanges.getNextUpdate(),
        "a change which was given back must not also be handed out as a dependency");

    // It is replayed by the delivery which takes it over, and by that one only.
    final ModifyDNMsg nextDelivery = renameIntoDeletedEntry(renamed);
    assertTrue(pendingChanges.putRemoteUpdate(nextDelivery));
    assertTrue(pendingChanges.markInProgress(nextDelivery));
    pendingChanges.commit(renamed);

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

  /**
   * Test case for [Issue 954]: the changes another thread parked are left alone.
   * <p>
   * A change is given back by the thread which owns it and by nobody else, here as
   * everywhere else (issue #922). The thread which parked a change is the one which may
   * still be inside the dependency checks which parked it - they park a change once per
   * dependency it has - so a change released under it would be listed as waiting again a
   * moment later, and handed out while the delivery which took it over is being replayed.
   */
  @Test
  public void theParkedChangesOfAnotherThreadAreLeftAlone() 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));

    // Another replay thread is unwound while this one holds the change it parked.
    runAndJoin(new Runnable()
    {
      @Override
      public void run()
      {
        assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), emptyList(),
            "a change another thread parked is not this one's to give back");
      }
    });

    assertEquals(pendingChanges.getDependentChangesSize(), 1,
        "the change must stay listed as waiting for the one it depends on");
    assertFalse(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
        "a change a replay thread owns must not be taken over (OPENDJ-1115)");

    // It is still handed to whichever thread clears the change it was waiting for.
    pendingChanges.commit(deleted);
    assertSame(pendingChanges.getNextUpdate(), rename);
  }

  /** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */
  private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception
  {

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

@@ -50,6 +50,7 @@
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -66,7 +67,9 @@
 * over that session, so a replay which fails while it is on its way must not restart it.
 * A restart asked for before the total update took the session, and left standing for the
 * length of it, is not run once it is over either: the change it was asked for is gone with
 * the ServerState the import replaced.
 * the ServerState the import replaced. The changes a replay which is unwound had parked as
 * waiting for another one are released on the same terms, and nothing more is done for them
 * (issue #954).
 * <p>
 * The exporter is a broker of this test, so that the test says when the entries arrive: the
 * change is replayed while the import is waiting for them - or, for the request, while the
@@ -293,6 +296,115 @@
  }

  /**
   * The changes a replay which is unwound had parked as waiting for another change are
   * released and nothing more while a total update owns the session (issue #954): no
   * session restart is asked for them - the one it would ask for is refused where it runs,
   * and the request would be spent on it - and they are neither reported as changes the
   * replication server sends again, which it does not before the import has replaced the
   * data, nor counted as processed. That is the road a change a stopping replay thread
   * abandons takes on this domain, and the give-back of the parked changes takes it too.
   * <p>
   * Pinned on the import road because it is the one road with an owner which a test holds
   * open for as long as it needs: the request is on its way until the exporter answers it,
   * and the backend is live meanwhile, so the change which is parked and the replay which
   * is unwound run as they would on any domain. The domain going away, or being disabled,
   * forgets its pending changes a moment after it takes the session and clears every
   * request and every count on its way, so a give-back on that road is a race with the
   * forgetting and leaves nothing to read.
   * <p>
   * The replay is unwound on the thread of this test - it applied its change, and the ack
   * of its delivery runs out of memory - so the parked change is this thread's to give
   * back, and the error which ends a replay thread is caught here instead.
   */
  @Test(timeOut = 120_000)
  public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain() throws Exception
  {
    final Entry entry = TestCaseUtils.addEntry(
        "dn: cn=renamedSince," + EXAMPLE_DN,
        "objectClass: top",
        "objectClass: person",
        "cn: renamedSince",
        "sn: renamedSince");
    final String entryUUID = getEntryUUID(entry.getName());
    final String[] exported = exportedEntries();

    // The request is out, and the exporter holds it until the give-back below has run.
    domain.initializeFromRemote(EXPORTER_ID, null);
    assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));

    /*
     * The barrier: a change whose replay fails stays listed and uncommitted - the attempts
     * in place end on an entryUUID search which does not run, the way they do in the case
     * above - and stays among the changes the newer ones are checked against, so a change
     * which follows it on the same entry has to wait for it. The restart which would have
     * followed is refused, the total update owning the session, and the search is let
     * through again before anything below reads a monitor.
     */
    final DN movedAway = DN.valueOf("cn=movedAway," + EXAMPLE_DN);
    final CSN failing = gen.newCSN();
    ShortCircuitPlugin.registerShortCircuit(
        OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
    try
    {
      replayMsg(new ModifyMsg(failing, movedAway,
          generatemods("description", "the replay of this change fails"), entryUUID));
    }
    finally
    {
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
    }
    assertFalse(domain.getServerState().cover(failing),
        "the change whose replay fails must stay listed as one which is not in the data");

    // Parked as waiting for it by this thread, which owns it from here on.
    final CSN parked = gen.newCSN();
    replayMsg(new ModifyMsg(parked, movedAway,
        generatemods("description", "the change which was parked as a dependency"), entryUUID));
    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
        "a change which waits for one that is not in the data must be parked");

    /*
     * The replay which is unwound while this thread still holds the parked change: its own
     * change is applied and committed, so the give-back on the way out finds the parked
     * change alone. The count is read once the parked change is listed, since a parked
     * change publishes no ack and is not counted until the delivery which replays it is.
     */
    final long processed = getMonitorAttrValue(baseDN, "replayed-updates");
    final CSN unwound = gen.newCSN();
    try
    {
      replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, entry.getName(),
          generatemods("description", "the replay of this change is unwound once it is applied"),
          entryUUID));
      Assert.fail("the replay was not unwound: the ack of the delivery must run out of memory");
    }
    catch (OutOfMemoryError unwinding)
    {
      // The error is the fixture's own, and this is the thread it would have ended.
    }

    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
        "the change parked by the replay which was unwound must be given back");
    assertEquals(getMonitorAttrValue(baseDN, "replayed-updates"), processed,
        "a change released while a total update owns the session must not be counted as"
            + " processed: no session sends it again before the import has replaced the data");
    assertThat(errorLogRecordsOf(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK.ordinal(), parked))
        .as("the change was reported as one the replication server sends again, which it does"
            + " not before the import has replaced the data")
        .isEmpty();
    assertTrue(domain.isConnected(), "the session the answer to the request arrives over was stopped");

    answerImportRequest(exported.length);
    finishImport(exported);
    for (String ldif : exported)
    {
      final DN dn = dnOf(ldif);
      assertTrue(entryExists(dn), "the import ended before " + dn
          + " arrived: the answer to the request was lost with the session it was made over");
    }
  }

  /**
   * A session restart which stood while the import ran was asked for by a replay thread
   * for a change given back before the total update owned the session, and that change is
   * forgotten with the pending changes when the imported data replaces the ServerState: