[#908] Wait for the changes being applied before a domain going down saves ...

Valery Kharseko

21 hours ago 661dc06886df4738b9add2206a39d74d7dedf476

[#908] Wait for the changes being applied before a domain going down saves its ServerState (#945)

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

@@ -57,6 +57,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import net.jcip.annotations.GuardedBy;

@@ -143,6 +144,7 @@
import org.opends.server.types.ExistingFileBehavior;
import org.opends.server.types.LDIFExportConfig;
import org.opends.server.types.LDIFImportConfig;
import org.opends.server.types.LockManager;
import org.opends.server.types.Modification;
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
@@ -389,6 +391,75 @@
  @GuardedBy("serviceStateLock")
  private long sessionGeneration;
  /**
   * Held while a replay thread applies a change of this domain, and taken exclusively by
   * this domain on its way down.
   * <p>
   * A change which reached the backend has to be recorded in the ServerState which is saved
   * when the domain is disabled or shut down, or it ends up in the data and in no
   * ServerState: the replication server sends it again and a change which is already
   * applied is replayed a second time (issue #908). The flags which stop the replay -
   * {@link #disabled}, {@link #shutdown} - are read under this lock as well, so a domain on
   * its way down sets one of them and then takes this lock: what it waits for is the
   * changes which were already being applied, and no attempt starts after that.
   * <p>
   * What this closes is the window between an operation reaching the backend and its
   * {@code commit()}, which is the one issue #908 reports. It is not every road by which a
   * change can be in the data and in no saved ServerState: {@code commit()} only advances
   * the state over the changes which are committed from the head of the pending list, so a
   * change applied while an older one is still to be replayed is forgotten by the
   * {@code clear()} on the way down and sent again - the barrier of issue #889 seen from
   * the other side, and a thing to fix where that barrier is rather than here. A change
   * this replica made itself is a third road, and the one the ServerState recovery in
   * {@code PersistentServerState.loadState()} already repairs, since it only looks for the
   * CSNs of this server.
   * <p>
   * Deliberately not the fair kind. The read lock is taken for every attempt made on the
   * backend, and a queued writer blocks the readers which come after it even without
   * fairness; the readers which do barge past it are the ones which see the flag and leave
   * without applying anything. The flag is read once before the lock for that reason too -
   * the replay threads are a pool shared by every domain, and a thread which parks on the
   * lock of a domain going down is a thread no other domain gets its changes replayed by.
   * <p>
   * The lock is released at the end of every attempt, so a domain going down waits for one
   * attempt and the short backoff which follows it inside the loop, rather than for all
   * the attempts a delivery is given. The backoff between two attempts is a Thread.sleep()
   * of tens of milliseconds; the one between two session restarts, which is counted in
   * seconds, is outside every lock and stays there.
   */
  private final ReentrantReadWriteLock replayLock = new ReentrantReadWriteLock();
  private final ReentrantReadWriteLock.ReadLock replayReadLock = replayLock.readLock();
  private final ReentrantReadWriteLock.WriteLock replayWriteLock = replayLock.writeLock();
  /**
   * How long this domain waits for the replay threads which are applying one of its changes
   * before it saves its ServerState and goes down.
   * <p>
   * Derived from the ceiling the server itself puts on an operation which is waiting for an
   * entry rather than picked: {@link LockManager} gives the subtree lock and the entry lock
   * {@link LockManager#DEFAULT_LOCK_TIMEOUT} each, so a replayed change whose target is held
   * by a concurrent local operation - a client deleting the subtree above it, say - is
   * inside its attempt for twice that before it gives up with BUSY. A bound under that
   * ceiling would be spent by ordinary lock contention, and the change which is applied
   * after it would be the one this whole barrier exists to keep out of that window: no
   * import, no index rebuild and no wedged backend needed.
   * <p>
   * A ceiling rather than a guarantee: an operation also waits for the subtree lock of every
   * entry above its target, one timeout each, so a deep contended chain outlasts this. What
   * it buys is that the wait is not lost to the contention a serving backend has anyway.
   * <p>
   * It is paid in three places - held under serviceStateLock, inside BackendConfigManager's
   * write lock when a backend is being deregistered, and once per domain by a server going
   * down - which is why it is bounded at all. It is only ever spent in full by a replay
   * which is genuinely stuck: the wait ends the moment the attempt does.
   */
  private static final long REPLAY_DRAIN_TIMEOUT_IN_MS =
      2 * LockManager.DEFAULT_LOCK_TIMEOUT_UNITS.toMillis(LockManager.DEFAULT_LOCK_TIMEOUT) + 1000;
  /**
   * How long this domain waits for the replay of its changes on its way down. Only the
   * tests, which can not hold a replay thread for {@link #REPLAY_DRAIN_TIMEOUT_IN_MS}, set
   * another value.
   */
  private volatile long replayDrainTimeoutInMs = REPLAY_DRAIN_TIMEOUT_IN_MS;
  /**
   * Stands for "the alert about a change this replica gave up on was never sent". The
   * time it is compared with only moves forward from an origin which is arbitrary, so
   * zero is not far enough in the past to say it.
@@ -556,7 +627,27 @@
          Thread.currentThread().interrupt();
        }
      }
      /*
       * A disabled domain saved its ServerState and cleared it from memory, and an import
       * or a restore is about to replace the data: saving here would write that empty state
       * over the saved one, which is a REPLACE of ds-sync-state with no value at all - the
       * replica would come back with no ServerState rather than with the one it saved on
       * its way down. The disabled flag does not catch the total update this replica is the
       * target of: preBackendImport() sets ignoreBackendInitializationEvent, so disable() is
       * not called on that road and only the import says the data is being replaced.
       *
       * The direction is asked for rather than ieRunning(), which the save in the loop above
       * settles for: an export leaves the data and the ServerState of this domain alone, and
       * the replay of this domain keeps running for the whole of it - a remote-requested
       * export is dispatched to a thread pool for that very reason. Since the save in the
       * loop is skipped for either direction, this is the only one which persists the
       * changes replayed since the export began, and there is no repair for them afterwards:
       * checkAndUpdateServerState() only repairs the CSNs of this server.
       */
      if (!disabled && !importInProgress())
      {
      state.save();
      }

      done = true;
    }
@@ -2355,6 +2446,15 @@
  {
    if (shutdown.compareAndSet(false, true))
    {
      /*
       * Wait for the changes which are being applied before the ServerState is flushed for
       * the last time: the flush thread stopped below is the last thing which saves it, so a
       * change which reaches the backend after that save is in the data and in no
       * ServerState (issue #908). The flag was just set, so this waits for the attempts
       * which had started already and no new one begins.
       */
      awaitReplayDrained();

      final RSUpdater rsUpdater = this.rsUpdater.get();
      if (rsUpdater != null)
      {
@@ -2475,31 +2575,35 @@
        int retryCount = IN_PLACE_REPLAY_ATTEMPTS;
        while (!dependency && !replayDone && retryCount-- > 0)
        {
          if (replayThreadShutdown.get() || shutdown.get() || disabled)
          /*
           * The flag which says this domain is going down is read before the lock as well
           * as under it. The replay threads are a pool shared by every domain of this
           * server, so a thread which took a change of a domain which is going down should
           * not queue behind the wait for that domain: the changes of every other domain
           * are behind it in the same pool.
           *
           * The read under the lock is the one which decides; the one above it is a
           * scheduling optimisation for the common case and nothing more. It cannot keep
           * this thread out of the queue: the flag can be set and the writer can queue
           * between the two reads, and a reader which arrives behind a queued writer blocks
           * even on a lock which is not the fair kind.
           */
          boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled;
          if (!goingDown)
          {
            /*
             * Either this replay thread or this domain is going away, or the domain is
             * being imported into or restored, so let's leave. The change was never
             * applied, so the ack says so - an assured write must not be told that a
             * change this replica is asking for again is in the data here - and the change
             * is given back to the replication server rather than left listed as being
             * replayed by a thread which is gone. Handing it back stops and starts the
             * session, so it waits until the ack has been published on the session this
             * delivery came over.
             *
             * A disabled domain saved its ServerState, cleared it from memory and forgot
             * its pending changes: a thread which kept applying changes into the backend
             * being imported into would have every one of its commits fail on a map which
             * is empty, one ERR_OPERATION_NOT_FOUND_IN_PENDING per change in flight, and
             * would be writing into a backend the import owns. abandonReplay() knows there
             * is nothing left to hand back in that case.
             * Every attempt made on the backend is under this lock, and so is the decision
             * to make one: a domain on its way down takes it exclusively once it has set
             * the flag read here, so a change which reaches the backend is recorded in the
             * ServerState which is saved on the way down, or is not applied at all
             * (issue #908).
             */
            replayErrorMsg =
                NOTE_REPLAY_ABANDONED_CHANGE.get(msg.getCSN(), getBaseDN()).toString();
            replayAbandoned = true;
            replayDone = true;
            break;
          }
            replayReadLock.lock();
            try
            {
              goingDown = replayThreadShutdown.get() || shutdown.get() || disabled;
              if (!goingDown)
              {
          if (!firstAttempt)
          {
            /*
@@ -2663,6 +2767,38 @@
            replayDone = true;
          }
        }
            }
            finally
            {
              replayReadLock.unlock();
            }
          }
          if (goingDown)
          {
            /*
             * Either this replay thread or this domain is going away, or the domain is
             * being imported into or restored, so let's leave. The change was never
             * applied, so the ack says so - an assured write must not be told that a
             * change this replica is asking for again is in the data here - and the change
             * is given back to the replication server rather than left listed as being
             * replayed by a thread which is gone. Handing it back stops and starts the
             * session, so it waits until the ack has been published on the session this
             * delivery came over.
             *
             * A disabled domain saved its ServerState, cleared it from memory and forgot
             * its pending changes: a thread which kept applying changes into the backend
             * being imported into would have every one of its commits fail on a map which
             * is empty, one ERR_OPERATION_NOT_FOUND_IN_PENDING per change in flight, and
             * would be writing into a backend the import owns. abandonReplay() knows there
             * is nothing left to hand back in that case.
             */
            replayErrorMsg =
                NOTE_REPLAY_ABANDONED_CHANGE.get(msg.getCSN(), getBaseDN()).toString();
            replayAbandoned = true;
            replayDone = true;
            break;
          }
        }

        if (!replayDone && !dependency)
        {
@@ -3855,11 +3991,29 @@
  {
    synchronized (serviceStateLock)
    {
      state.save();
      state.clearInMemory();
      /*
       * The replay is stopped before the ServerState is saved, and not the other way round:
       * a change a replay thread is applying has to be either recorded in the state which
       * is about to be saved or not applied at all, or it ends up in the data and in no
       * ServerState (issue #908). The flag keeps the attempts which have not started from
       * starting - it is read under the same lock as the attempt it guards - and the wait
       * below is for the ones which had started already.
       *
       * All of it stays under serviceStateLock, so that this and enable() remain the
       * mutually exclusive pair they have always been: an enable() which ran in the middle
       * of this would clear the flag and bring a session up, and this would then go on to
       * cut that session and clear a ServerState which the flush thread - reading a flag
       * which says the domain is enabled - would write back empty. That is what bounds
       * REPLAY_DRAIN_TIMEOUT_IN_MS: the wait is held under a lock which a session restart,
       * a configuration change and the shutdown of this domain take, and it runs inside
       * BackendConfigManager's write lock when a backend is being deregistered.
       */
      disabled = true;
      disableService(); // This will cut the session and wake up the listener
      sessionGeneration++;
      awaitReplayDrained();
      state.save();
      state.clearInMemory();
      /*
       * The ServerState this bookkeeping goes with is now gone from memory and is loaded
       * again from the backend when the domain is enabled back, so the changes listed as
@@ -3880,6 +4034,90 @@
  }

  /**
   * Waits for the replay threads which are applying a change of this domain to be done
   * with it.
   * <p>
   * Called once {@link #disabled} or {@link #shutdown} has been set, which is what bounds
   * the wait: a replay thread reads those under {@link #replayReadLock}, the lock this
   * takes exclusively, so no attempt starts once this returns and what it waits for is the
   * attempts which were running already. The lock is released before returning for the
   * same reason - what keeps the replay out is the flag, not the lock.
   */
  private void awaitReplayDrained()
  {
    boolean drained = false;
    boolean interrupted = false;
    try
    {
      drained = replayWriteLock.tryLock(replayDrainTimeoutInMs, TimeUnit.MILLISECONDS);
    }
    catch (InterruptedException e)
    {
      /*
       * Give up waiting, and put the interrupt back rather than swallow it: whoever
       * interrupted this thread - the server going down, a thread pool taking its threads
       * away - is still waiting for it to stop, and this is not the last thing it does.
       * The cost is on the shutdown road, whose wait for the last ServerState flush is a
       * Thread.sleep() which ends on its first call once the flag is set: the flush thread
       * still runs that save, this one just stops waiting for it.
       */
      interrupted = true;
      Thread.currentThread().interrupt();
    }
    if (drained)
    {
      replayWriteLock.unlock();
      return;
    }
    /*
     * The change which is being applied may reach the backend without being recorded in the
     * ServerState which is saved next, so the replication server sends it again and it is
     * replayed a second time. Better than holding an administrative task - an import, a
     * restore, a backend being taken offline - for as long as a backend which stopped
     * answering takes to answer.
     *
     * An interrupted wait is reported as what it is: it says nothing about how long the
     * replay of this domain takes, and the timeout it never spent would have an operator
     * reading a backend which is slow into it.
     */
    if (interrupted)
    {
      logger.warn(WARN_REPLAY_DRAIN_INTERRUPTED, getBaseDN());
    }
    else
    {
      logger.warn(WARN_REPLAY_NOT_DRAINED, getBaseDN(), replayDrainTimeoutInMs);
    }
  }

  /**
   * Returns how long this domain waits for the replay threads which are applying one of
   * its changes before it saves its ServerState and goes down.
   *
   * @return the timeout in milliseconds
   */
  @VisibleForTesting
  public long getReplayDrainTimeout()
  {
    return replayDrainTimeoutInMs;
  }

  /**
   * Sets how long this domain waits for the replay threads which are applying one of its
   * changes before it saves its ServerState and goes down.
   * <p>
   * Only there for the tests which check what a domain does when that wait runs out: they
   * can not hold a replay thread for {@link #REPLAY_DRAIN_TIMEOUT_IN_MS}.
   *
   * @param timeoutInMs the timeout in milliseconds
   */
  @VisibleForTesting
  public void setReplayDrainTimeout(long timeoutInMs)
  {
    replayDrainTimeoutInMs = timeoutInMs;
  }

  /**
   * Do what necessary when the data have changed : load state, load
   * generation Id.
   * If there is no such information check if there is a

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

@@ -2736,6 +2736,24 @@
  }

  /**
   * Returns a boolean indicating if a total update <em>into</em> this replica is currently
   * processed, that is an import which is replacing the data of this domain.
   * <p>
   * The other direction, an export which is initializing another replica from this one,
   * leaves the data of this domain and its ServerState alone: it is reported by
   * {@link #ieRunning()} just the same, so anything which is guarding against the data
   * being replaced has to ask this rather than that.
   *
   * @return {@code true} when an import is being processed, {@code false} when nothing is
   *         or when what is being processed is an export
   */
  protected boolean importInProgress()
  {
    final ImportExportContext ieCtx = importExportContext.get();
    return ieCtx != null && ieCtx.importInProgress();
  }

  /**
   * Check the value of the Replication Servers generation ID.
   *
   * @param generationID        The expected value of the generation ID.

 opendj-server-legacy/src/main/java/org/opends/server/types/LockManager.java

@@ -12,6 +12,7 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2015-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.types;

@@ -256,8 +257,19 @@
    }
  }

  private static final long DEFAULT_LOCK_TIMEOUT = 9;
  private static final TimeUnit DEFAULT_LOCK_TIMEOUT_UNITS = TimeUnit.SECONDS;
  /**
   * How long a lock manager created with the default configuration - which is the one the
   * server runs with - waits for each of the locks an operation needs before it gives up
   * on the entry. An operation is made to wait for it more than once: the subtree lock and
   * the entry lock are taken one after the other, each with this timeout of its own.
   *
   * @see #DEFAULT_LOCK_TIMEOUT_UNITS
   */
  public static final long DEFAULT_LOCK_TIMEOUT = 9;
  /**
   * The unit of {@link #DEFAULT_LOCK_TIMEOUT}.
   */
  public static final TimeUnit DEFAULT_LOCK_TIMEOUT_UNITS = TimeUnit.SECONDS;
  private static final int MINIMUM_NUMBER_OF_BUCKETS = 64;
  private static final int THREAD_LOCAL_CACHE_SIZE = 8;


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

@@ -626,3 +626,11 @@
NOTE_REPLAY_ABANDONED_CHANGE_309=Could not replay change %s in domain "%s": the replay thread \
 it was given to is stopping. The change has not been recorded as replayed and is given back to \
 the replication server, which still owns it
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 \
 and it is replayed a second time
WARN_REPLAY_DRAIN_INTERRUPTED_320=Domain "%s" is going down and was interrupted while it waited \
 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 and it \
 is replayed a second time

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

New file
@@ -0,0 +1,305 @@
/*
 * 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.plugins;

import static java.util.concurrent.TimeUnit.*;

import java.util.ArrayList;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizedIllegalArgumentException;
import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.server.PluginCfg;
import org.opends.server.api.plugin.DirectoryServerPlugin;
import org.opends.server.api.plugin.PluginResult;
import org.opends.server.api.plugin.PluginType;
import org.opends.server.types.OperationType;
import org.opends.server.types.operation.PluginOperation;
import org.opends.server.types.operation.PreParseAddOperation;
import org.opends.server.types.operation.PreParseDeleteOperation;
import org.opends.server.types.operation.PreParseModifyDNOperation;
import org.opends.server.types.operation.PreParseModifyOperation;

/**
 * A plugin which parks an operation at the pre-parse plugin point until the test releases
 * it, so that a test can hold a thread inside {@code Operation.run()} for as long as it
 * takes to do something else.
 * <p>
 * The pre-parse point is the one the replayed operations of a replication domain go
 * through - the pre-operation plugins are not invoked for synchronization operations - so
 * this is how a test holds a replay thread between the moment it starts applying a change
 * and the moment the change reaches the backend.
 * <p>
 * A pause is registered for one operation type on one entry: this plugin is enabled for
 * the whole unit test suite and for the internal operations too, so a pause which parked
 * every operation of a type would park whatever else the server happens to be doing - and
 * report that as the operation the test is waiting for. The park is bounded all the same:
 * a test which never releases costs {@link #MAX_PAUSE_IN_MS} rather than a server thread.
 */
public class PausePreParsePlugin extends DirectoryServerPlugin<PluginCfg>
{
  /**
   * How long an operation is parked when nothing releases it. A test which forgets to
   * release, or which fails before it could, must not leave a server thread parked for
   * the rest of the run.
   */
  public static final long MAX_PAUSE_IN_MS = 60000;

  /** One registered pause: the entry it applies to, and what reports and releases it. */
  private static final class Pause
  {
    /** The entry whose operations are parked. */
    private final DN target;
    /** Counted down by the first operation which reaches the pause. */
    private final CountDownLatch reached = new CountDownLatch(1);
    /** Counted down when the test releases the parked operations. */
    private final CountDownLatch released = new CountDownLatch(1);
    /** How many operations are parked here right now. */
    private final AtomicInteger parked = new AtomicInteger();

    private Pause(DN target)
    {
      this.target = target;
    }
  }

  /** The pauses registered per operation type. */
  private static final Map<OperationType, Pause> pauses = new ConcurrentHashMap<>();

  /**
   * Creates a new instance of this Directory Server plugin. Every plugin must implement a
   * default constructor (it is the only one that will be used to create plugins defined in
   * the configuration), and every plugin constructor must call <CODE>super()</CODE> as its
   * first element.
   */
  public PausePreParsePlugin()
  {
    super();
  }

  @Override
  public void initializePlugin(Set<PluginType> pluginTypes, PluginCfg configuration) throws ConfigException
  {
    // This plugin may only be used as a pre-parse plugin.
    for (PluginType t : pluginTypes)
    {
      switch (t)
      {
        case PRE_PARSE_ADD:
        case PRE_PARSE_DELETE:
        case PRE_PARSE_MODIFY:
        case PRE_PARSE_MODIFY_DN:
          // This is fine.
          break;
        default:
          throw new ConfigException(
              LocalizableMessage.raw("Invalid plugin type " + t + " for the pause pre-parse plugin."));
      }
    }
  }

  @Override
  public void finalizePlugin()
  {
    /*
     * A pause which outlived the test which registered it - one whose test timed out
     * before its finally could run - must not park the operations of the tests which
     * follow. This runs when the plugin is taken away, which is when the test server is
     * stopped or restarted.
     */
    for (OperationType operation : new ArrayList<>(pauses.keySet()))
    {
      release(operation);
    }
  }

  @Override
  public PluginResult.PreParse doPreParse(PreParseAddOperation addOperation)
  {
    pauseInternal(addOperation, addOperation.getRawEntryDN());
    return PluginResult.PreParse.continueOperationProcessing();
  }

  @Override
  public PluginResult.PreParse doPreParse(PreParseDeleteOperation deleteOperation)
  {
    pauseInternal(deleteOperation, deleteOperation.getRawEntryDN());
    return PluginResult.PreParse.continueOperationProcessing();
  }

  @Override
  public PluginResult.PreParse doPreParse(PreParseModifyOperation modifyOperation)
  {
    pauseInternal(modifyOperation, modifyOperation.getRawEntryDN());
    return PluginResult.PreParse.continueOperationProcessing();
  }

  @Override
  public PluginResult.PreParse doPreParse(PreParseModifyDNOperation modifyDNOperation)
  {
    pauseInternal(modifyDNOperation, modifyDNOperation.getRawEntryDN());
    return PluginResult.PreParse.continueOperationProcessing();
  }

  /**
   * Parks the operation if a pause is registered for its type and its entry, and reports
   * that it reached the pause so that the test knows the thread is now inside
   * {@code Operation.run()}.
   *
   * @param operation the operation which is being processed
   * @param rawEntryDN the entry it is on, as the request carries it
   */
  private void pauseInternal(PluginOperation operation, ByteString rawEntryDN)
  {
    final Pause pause = pauses.get(operation.getOperationType());
    if (pause == null || !pause.target.equals(parseOrNull(rawEntryDN)))
    {
      return;
    }
    /*
      * Counted before the pause is reported as reached, not after: a test which is told
      * that an operation got here goes on to read parkedCount(), and a count incremented
      * afterwards would leave that read racing this thread being scheduled.
      */
    pause.parked.incrementAndGet();
    pause.reached.countDown();
    try
    {
      pause.released.await(MAX_PAUSE_IN_MS, MILLISECONDS);
    }
    catch (InterruptedException e)
    {
      Thread.currentThread().interrupt();
    }
    finally
    {
      pause.parked.decrementAndGet();
    }
  }

  /** Returns the DN the request carries, or {@code null} when it does not parse. */
  private static DN parseOrNull(ByteString rawEntryDN)
  {
    if (rawEntryDN == null)
    {
      return null;
    }
    try
    {
      return DN.valueOf(rawEntryDN.toString());
    }
    catch (LocalizedIllegalArgumentException e)
    {
      // Not the entry any pause is registered for, whatever it was meant to be.
      return null;
    }
  }

  /**
   * Registers a pause: the operations of the given type on the given entry are parked at
   * the pre-parse plugin point until {@link #release(OperationType)} is called.
   *
   * @param operation the type of operation to park
   * @param target the entry whose operations are parked
   */
  public static void pause(OperationType operation, DN target)
  {
    final Pause replaced = pauses.put(operation, new Pause(target));
    if (replaced != null)
    {
      // Whatever was parked on the pause this one replaces would stay parked for the rest
      // of MAX_PAUSE_IN_MS: the thread which registered it is not coming back for it.
      replaced.released.countDown();
    }
  }

  /**
   * Waits until an operation reached the pause registered for the given operation type.
   *
   * @param operation the type of operation which was registered
   * @param timeout how long to wait
   * @param unit the unit of the timeout
   * @return {@code true} when an operation reached the pause, {@code false} when the wait
   *         timed out
   * @throws IllegalStateException when no pause is registered for that operation type,
   *         which is a caller waiting for something nothing can report rather than an
   *         operation which is slow to come
   * @throws InterruptedException when the wait was interrupted
   */
  public static boolean awaitPaused(OperationType operation, long timeout, TimeUnit unit)
      throws InterruptedException
  {
    final Pause pause = pauses.get(operation);
    if (pause == null)
    {
      /*
       * Told apart from the timeout, and loudly: a pause registered for another operation
       * type - the whole of the mistake - would otherwise be reported as the operation
       * never coming, after the caller waited its whole budget out for it.
       */
      throw new IllegalStateException(
          "no pause is registered for " + operation + ": nothing can park on it, and nothing"
              + " will report that it did");
    }
    return pause.reached.await(timeout, unit);
  }

  /**
   * Returns how many operations are parked right now by the pause registered for the given
   * operation type.
   * <p>
   * A test which took something down while an operation was parked reads this to say that
   * it really did come down without waiting for it: the pause is only released by the test
   * itself, so an operation which is still parked here never finished.
   * <p>
   * To be read before the pause is released, and not after: the count is decremented by the
   * parked thread itself, on its way out, so a release does not bring it back to zero by the
   * time it returns.
   *
   * @param operation the type of operation which was registered
   * @return the number of operations parked right now, 0 when no pause is registered
   */
  public static int parkedCount(OperationType operation)
  {
    final Pause pause = pauses.get(operation);
    return pause != null ? pause.parked.get() : 0;
  }

  /**
   * Releases the operations parked by the pause registered for the given operation type,
   * and deregisters it so that the operations which follow are not parked. Does nothing
   * when no pause is registered, so that a test can release in a {@code finally} whatever
   * happened.
   *
   * @param operation the type of operation which was registered
   */
  public static void release(OperationType operation)
  {
    final Pause pause = pauses.remove(operation);
    if (pause != null)
    {
      pause.released.countDown();
    }
  }
}

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

@@ -54,12 +54,14 @@
import org.opends.server.core.ModifyOperation;
import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.extensions.DummyAlertHandler;
import org.opends.server.plugins.PausePreParsePlugin;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.plugins.ShortCircuitPlugin.ParkedReplay;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.LDAPReplicationDomain;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.AckMsg;
@@ -109,6 +111,45 @@
  /** The configuration attribute which carries the replay give-up budget of a domain. */
  private static final String ATTR_REPLAY_GIVE_UP_DELAY = "ds-cfg-replay-give-up-delay";

  /**
   * How long a replay parked by {@link PausePreParsePlugin} is held after the domain cut its
   * session, in the test which checks that a change being applied is recorded in the
   * ServerState a domain going down saves.
   * <p>
   * It has to be long enough for a domain which does not wait for the replay to have saved
   * its ServerState by the time the change is applied - that is the failure the test
   * reports - and well under the time a domain which does wait gives the replay, which that
   * test leaves at its default. Spent inside that wait, so what it costs the test is itself
   * and nothing more.
   * <p>
   * Counted from the moment the session was cut rather than from the wait, because that is
   * the only moment this test can see: {@code ReplicationBroker.stop()} is the first
   * statement of {@code disableService()}, and it stops the domain being connected before
   * the listener thread is asked to stop and joined - a join with no bound on it - and
   * before the ServerState is saved. So what this delay has to outlast is that whole
   * remainder of {@code disable()} and not the save alone. The remainder is a millisecond
   * on an idle machine and hundreds of them on a loaded one, and a delay of the same order
   * would hand the released replay a race against the save rather than a loss to it: the
   * change would be recorded whether or not anything waited for it, and the test would stop
   * saying anything without ever failing.
   */
  private static final long SETTLE_BEFORE_RELEASE_IN_MS = 2000;

  /**
   * How long a domain is told to wait for a replay it can not drain, in the test which
   * checks that it gives up rather than hold the task which is taking it down. Long enough
   * to be told apart from not waiting at all, short enough for a test to spend.
   * <p>
   * Told apart from the rest of {@code disable()}, to be exact: the test times that call as
   * a whole rather than the wait inside it, so this budget is what has to dominate cutting
   * the session, joining the listener thread with no bound on the join, and saving the
   * ServerState with an internal modify. That remainder is a millisecond on an idle machine
   * and hundreds of them on a loaded one, and a budget of the same order would have the
   * assertion satisfied by the remainder alone - the wait taken out of the domain and
   * nothing reporting it.
   */
  private static final long TEST_REPLAY_DRAIN_TIMEOUT_IN_MS = 2000;

  /** An entry with a entryUUID. */
  private Entry personWithUUIDEntry;
  private Entry personWithSecondUniqueID;
@@ -2613,6 +2654,338 @@
  }

  /**
   * Test case for [Issue 908]: a domain being disabled - for an LDIF import, a restore, or
   * a backend being taken offline - must not save its ServerState while a replay thread is
   * half way through applying one of its changes.
   * <p>
   * The change reaches the backend, so a ServerState which excludes it records nowhere
   * that it was applied: the replication server sends it again when the domain is enabled
   * back, and a change which is already in the data is replayed a second time - resolved
   * as a conflict, or left as a conflict entry when the changes around it were resent with
   * it and their dependency ordering was forgotten along with the pending changes.
   */
  @Test
  public void aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled() throws Exception
  {
    testSetUp("aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled"));

    final int serverId = 19;
    ReplicationBroker broker =
        openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
    try
    {
      final CSNGenerator gen = new CSNGenerator(serverId, 0);

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

      final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
      final CSN csn = gen.newCSN();
      boolean disableAttempted = false;
      try
      {
        /*
         * Park the replay inside op.run(): the pre-parse plugin point is reached once the
         * replay thread started applying the change and before the change reaches the
         * backend, which is the window this issue is about. The pre-operation point would
         * not do - it is not invoked for synchronization operations.
         */
        PausePreParsePlugin.pause(OperationType.DELETE, dn);
        broker.publish(new DeleteMsg(dn, csn, uuid));
        assertTrue(PausePreParsePlugin.awaitPaused(OperationType.DELETE, 60, SECONDS),
            "the replay thread never started applying the change");
        assertTrue(domain.isConnected(),
            "this test needs a domain which is still up when the change is being applied");

        /*
         * Let the parked replay finish once the domain is inside the wait for it, so that
         * the change reaches the backend while the ServerState is about to be saved. The
         * session is cut after the flag is set and immediately before that wait, and well
         * before the state is saved, so a domain which is not connected anymore is one which
         * is about to wait for this very change.
         *
         * Released a moment after that rather than on the disconnection itself, and this is
         * what makes the test decide rather than guess: a domain which does not wait - the
         * lock taken out of the replay, or the state saved before the wait as it was before
         * this fix - has saved its ServerState long before the delay is out, so the change
         * lands after that save and the assertion below reports it. Releasing on the
         * disconnection instead handed the replay the join of the listener thread as a head
         * start, which is enough for it to be recorded by a domain which never waited.
         *
         * The delay is spent inside the wait, so it costs this test nothing and holds
         * whatever budget it needs to be well under REPLAY_DRAIN_TIMEOUT_IN_MS.
         */
        final Thread releaser =
            releaseWhenDisconnected(domain, OperationType.DELETE, SETTLE_BEFORE_RELEASE_IN_MS);
        disableAttempted = true;
        try
        {
          domain.disable();
        }
        finally
        {
          releaser.join(SECONDS.toMillis(60));
        }

        /*
         * The entry is gone, so the change did reach the backend: getEntry() waits for it
         * and reports it, since a domain which did not wait for the replay lets it finish
         * a moment later rather than not at all.
         */
        getEntry(dn, 30000, false);
        /*
         * Read the ServerState which was saved rather than the one in memory: disable()
         * clears the in-memory one, and the saved one is what the domain reads back when
         * it is enabled again - and what the replication server resumes this replica from.
         * Read it before the domain is enabled back, or the change being sent again and
         * replayed a second time would make the state cover it either way, which is the
         * very outcome this test is about.
         */
        assertTrue(persistedServerState().cover(csn),
            "a change which reached the backend must be recorded in the saved ServerState");
      }
      finally
      {
        PausePreParsePlugin.release(OperationType.DELETE);
        if (disableAttempted)
        {
          /*
           * Only when disable() was reached, and whether or not it got to the end: setting
           * the flag is its first act, so a disable() which threw half way through still
           * left a domain which has to be enabled back. Enabling one which was never
           * disabled is what must not happen - it would reload the ServerState and start a
           * broker which is already running, behind the back of the tests which follow.
           */
          domain.enable();
        }
      }
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Test case for [Issue 908]: a domain which can not get the replay of its changes to
   * finish goes down anyway rather than holding the administrative task which is taking it
   * down - an import, a restore, a backend being taken offline - for as long as a backend
   * which stopped answering takes to answer.
   * <p>
   * The change may then reach the backend without being recorded in the ServerState, which
   * is what the warning in the log says: the replication server sends it again once the
   * domain is enabled back, which this test also checks, since a domain which gave up on
   * the wait must still end up consistent.
   */
  @Test
  public void theDomainStopsWaitingForAReplayWhichDoesNotFinish() throws Exception
  {
    testSetUp("theDomainStopsWaitingForAReplayWhichDoesNotFinish");
    logger.error(LocalizableMessage.raw(
        "Starting replication test : theDomainStopsWaitingForAReplayWhichDoesNotFinish"));

    final int serverId = 20;
    ReplicationBroker broker =
        openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
    try
    {
      final CSNGenerator gen = new CSNGenerator(serverId, 0);

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

      final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
      final CSN csn = gen.newCSN();
      final long drainTimeout = domain.getReplayDrainTimeout();
      boolean disableAttempted = false;
      try
      {
        /*
         * A replay which does not finish is waited out for as long as an operation can be
         * waiting for the entry it is on - the best part of twenty seconds: this test can
         * not, so the domain gives up on the wait after a moment instead. Set inside the
         * try which puts it back, like the pause below: both are the domain's and the
         * server's for as long as they are left behind.
         */
        domain.setReplayDrainTimeout(TEST_REPLAY_DRAIN_TIMEOUT_IN_MS);
        PausePreParsePlugin.pause(OperationType.DELETE, dn);
        broker.publish(new DeleteMsg(dn, csn, uuid));
        assertTrue(PausePreParsePlugin.awaitPaused(OperationType.DELETE, 60, SECONDS),
            "the replay thread never started applying the change");

        // The replay is parked and stays parked: the domain has to come down all the same.
        final long startedAt = System.nanoTime();
        disableAttempted = true;
        domain.disable();
        final long waitedMs = NANOSECONDS.toMillis(System.nanoTime() - startedAt);
        /*
         * Only this test releases the pause, and it has not done so yet, so an operation
         * still parked here is one the domain came down without waiting for - which is what
         * the give-up is. Read before the finally below releases it.
         */
        Assertions.assertThat(PausePreParsePlugin.parkedCount(OperationType.DELETE))
            .as("the domain must have come down while the replay was still being applied")
            .isEqualTo(1);
        /*
          * Measured against the default this test overrode rather than against a copy of
          * its value: an override which stopped taking effect would have the domain wait
          * the whole default out, and that is what this has to catch.
          */
        assertTrue(waitedMs < drainTimeout,
            "the domain waited " + waitedMs + " ms for a replay it can not drain,"
                + " which is not short of the " + drainTimeout + " ms it waits by default");
        /*
         * And the wait was taken rather than skipped: the replay is parked for good, so a
         * domain which really waits for it spends the whole budget it was given. Without
         * this the test reports the same thing whether the domain waited for the changes in
         * flight or never waited for anything - the give-up is only half of what a bounded
         * wait is.
         */
        assertTrue(waitedMs >= TEST_REPLAY_DRAIN_TIMEOUT_IN_MS,
            "the domain came down in " + waitedMs + " ms, so it did not wait the "
                + TEST_REPLAY_DRAIN_TIMEOUT_IN_MS + " ms it was given for the replay of a"
                + " change which was still being applied");
      }
      finally
      {
        PausePreParsePlugin.release(OperationType.DELETE);
        domain.setReplayDrainTimeout(drainTimeout);
        if (disableAttempted)
        {
          domain.enable();
        }
      }

      /*
       * The entry goes away: the replay the domain gave up on was released by the finally
       * above and finished after the ServerState had been saved, which is what the give-up
       * costs. This says the change is in the data - not that it was delivered again, since
       * the delete which does it is the first replay rather than the second.
       */
      getEntry(dn, 30000, false);
      /*
       * The change is in the data and in no ServerState, so the replication server owns it
       * still and sends it again over the session which the domain being enabled back
       * brought up. Replaying it a second time is the cost of the wait running out, and
       * conflict resolution absorbs it - what must not happen is the replica staying behind
       * for good. The state coming to cover the CSN is what evidences that delivery: the
       * domain forgot the change with its pending changes, so nothing else records it.
       */
      TestTimer timer = new TestTimer.Builder()
        .maxSleep(60, SECONDS)
        .sleepTimes(200, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          assertTrue(domain.getServerState().cover(csn),
              "the change must be recorded once it has been delivered again");
        }
      });
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Starts a thread which releases the operations parked by
   * {@link PausePreParsePlugin} once the domain has cut its session - which it does on its
   * way down, immediately before it waits for the replay of the changes in flight - plus a
   * delay which puts the release inside that wait rather than ahead of it.
   *
   * @param domain the domain which is about to be taken down
   * @param operation the type of operation the pause was registered for
   * @param settleInMs how long to wait after the session was cut before the parked
   *                   operations are released, which has to be well under the time the
   *                   domain waits for them and longer than the rest of {@code disable()} -
   *                   the session being cut is its first act, so the listener thread being
   *                   joined and the ServerState being saved are both inside this delay
   * @return the thread, already started
   */
  private Thread releaseWhenDisconnected(
      final LDAPReplicationDomain domain, final OperationType operation, final long settleInMs)
  {
    final Thread releaser = new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        /*
         * Bounded, and a daemon: a domain which never goes down - because taking it down
         * threw - must not leave a thread spinning for the rest of the run. The pause has
         * a bound of its own, so the parked operation is released either way.
         */
        final long deadline = System.nanoTime() + SECONDS.toNanos(60);
        try
        {
          while (domain.isConnected() && System.nanoTime() - deadline < 0)
          {
            Thread.sleep(1);
          }
          /*
           * The session is cut, so the domain is on its way to the wait for the replay:
           * give it that long to get there and, if it is not waiting for anything, to save
           * the ServerState this change must be in.
           */
          Thread.sleep(settleInMs);
          PausePreParsePlugin.release(operation);
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
        }
      }
    }, "issue 908 replay releaser");
    releaser.setDaemon(true);
    releaser.start();
    return releaser;
  }

  /**
   * Returns the ServerState of the test domain as it is saved in the backend.
   * <p>
   * That is the one the domain reads back when it is enabled again, and the one the
   * replication server resumes this replica from - the in-memory one is cleared by
   * {@code disable()}.
   *
   * @return the ServerState read from the base entry of the domain
   * @throws Exception if the base entry could not be read
   */
  private ServerState persistedServerState() throws Exception
  {
    final ServerState persisted = new ServerState();
    for (String value : getEntry(baseDN, 1, true).parseAttribute("ds-sync-state").asSetOfString())
    {
      persisted.update(new CSN(value));
    }
    return persisted;
  }

  /**
   * A ModifyMsg whose operation can not tell which change it carries.
   * <p>
   * The operation is built - so the replay is past the point where a message is given up

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

@@ -13,7 +13,7 @@
 *
 * Copyright 2008-2010 Sun Microsystems, Inc.
 * Portions Copyright 2013-2016 ForgeRock AS.
 * Portions Copyright 2025 3A Systems LLC.
 * Portions Copyright 2025-2026 3A Systems LLC.
 */
package org.opends.server.replication.service;

@@ -46,6 +46,16 @@
  /** A StringBuffer that will be used to build a new String should the import be called. */
  private StringBuffer importString;
  private int exportedEntryCount;
  /**
   * What {@code importInProgress()} said while this domain was inside
   * {@link #importBackend(InputStream)}, and {@code null} when it was never there.
   */
  private volatile Boolean importReportedDuringImport;
  /**
   * What {@code importInProgress()} said while this domain was inside
   * {@link #exportBackend(OutputStream)}, and {@code null} when it was never there.
   */
  private volatile Boolean importReportedDuringExport;

  private FakeReplicationDomain(DN baseDN, int serverID,
      SortedSet<String> replicationServers, int window, long heartbeatInterval,
@@ -104,9 +114,28 @@
    return exportedEntryCount;
  }

  /**
   * Returns what {@code importInProgress()} said while this domain was inside
   * {@code importBackend()}, or {@code null} when it never was.
   */
  Boolean importReportedDuringImport()
  {
    return importReportedDuringImport;
  }

  /**
   * Returns what {@code importInProgress()} said while this domain was inside
   * {@code exportBackend()}, or {@code null} when it never was.
   */
  Boolean importReportedDuringExport()
  {
    return importReportedDuringExport;
  }

  @Override
  protected void exportBackend(OutputStream output) throws DirectoryException
  {
    importReportedDuringExport = importInProgress();
    try
    {
      output.write(exportString.getBytes());
@@ -122,6 +151,7 @@
  @Override
  protected void importBackend(InputStream input) throws DirectoryException
  {
    importReportedDuringImport = importInProgress();
    byte[] buffer = new byte[1000];
    int ret;
    do

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

@@ -782,13 +782,24 @@
    }
  }

  private void assertExportSucessful(ReplicationDomain domain1,
      ReplicationDomain domain2, String exportedData, StringBuffer importedData)
  private void assertExportSucessful(FakeReplicationDomain domain1,
      FakeReplicationDomain domain2, String exportedData, StringBuffer importedData)
  {
    assertEquals(getLeftEntryCount(domain2), 0, "Wrong LeftEntryCount for export");
    assertEquals(getLeftEntryCount(domain1), 0, "Wrong LeftEntryCount for import");
    assertEquals(importedData.length(), exportedData.length());
    assertEquals(importedData.toString(), exportedData);
    /*
     * The direction of the total update, as each side reported it while it was in it.
     * ieRunning() cannot tell the two apart, and something which is guarding the data of a
     * replica has to: the exporter keeps its data and its ServerState - and keeps replaying
     * into them - while the importer is having both replaced.
     */
    assertEquals(domain2.importReportedDuringImport(), Boolean.TRUE,
        "the replica whose data is being replaced must report a total update into itself");
    assertEquals(domain1.importReportedDuringExport(), Boolean.FALSE,
        "the replica which is exporting must not report a total update into itself:"
            + " its data and its ServerState are left alone");
  }

  private long getLeftEntryCount(ReplicationDomain domain)

 opendj-server-legacy/tests/unit-tests-testng/resource/config-changes.ldif

@@ -331,6 +331,19 @@
ds-cfg-plugin-type: preOperationSearch
ds-cfg-invoke-for-internal-operations: true

dn: cn=Pause PreParse Plugin,cn=Plugins,cn=config
changetype: add
objectClass: top
objectClass: ds-cfg-plugin
cn: Pause PreParse Plugin
ds-cfg-java-class: org.opends.server.plugins.PausePreParsePlugin
ds-cfg-enabled: true
ds-cfg-plugin-type: preParseAdd
ds-cfg-plugin-type: preParseDelete
ds-cfg-plugin-type: preParseModify
ds-cfg-plugin-type: preParseModifyDN
ds-cfg-invoke-for-internal-operations: true

dn: cn=Update PreOperation Plugin,cn=Plugins,cn=config
changetype: add
objectClass: top