From 661dc06886df4738b9add2206a39d74d7dedf476 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 10 Sep 2026 11:57:37 +0000
Subject: [PATCH] [#908] Wait for the changes being applied before a domain going down saves its ServerState (#945)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 15
opendj-server-legacy/src/main/java/org/opends/server/types/LockManager.java | 16
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 8
opendj-server-legacy/src/test/java/org/opends/server/plugins/PausePreParsePlugin.java | 305 +++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 373 ++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 570 +++++++++++++++++-------
opendj-server-legacy/src/test/java/org/opends/server/replication/service/FakeReplicationDomain.java | 32 +
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 18
opendj-server-legacy/tests/unit-tests-testng/resource/config-changes.ldif | 13
9 files changed, 1,179 insertions(+), 171 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index add2d58..4ad0c39 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
+++ b/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();
}
}
- state.save();
+ /*
+ * 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,7 +2575,205 @@
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)
+ {
+ /*
+ * 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).
+ */
+ replayReadLock.lock();
+ try
+ {
+ goingDown = replayThreadShutdown.get() || shutdown.get() || disabled;
+ if (!goingDown)
+ {
+ if (!firstAttempt)
+ {
+ /*
+ * Every attempt runs an operation of its own. An Operation which already ran
+ * carries the request controls and the access log items of that run, so
+ * re-running the same one stacks one ManageDsaIT control - and one access log
+ * record - per attempt. It also picks up the new state of the UpdateMsg when
+ * conflict resolution rewrote it.
+ * Note: When msg is a DeleteMsg, the DeleteOperation is properly created
+ * with subtreeDelete request control when needed.
+ */
+ nextOp = msg.createOperation(conn);
+ }
+ firstAttempt = false;
+
+ // Try replay the operation
+ op = nextOp;
+ op.setInternalOperation(true);
+ op.setSynchronizationOperation(true);
+
+ // Always add the ManageDSAIT control so that updates to referrals
+ // are processed locally.
+ op.addRequestControl(new LDAPControl(OID_MANAGE_DSAIT_CONTROL));
+
+ // Warning: specific processing ahead. See OPENDJ-2792
+ if (op instanceof ModifyOperation)
+ {
+ ModifyOperation modifyOperation = (ModifyOperation) op;
+ if (modifyOperation.getEntryDN().equals(SET_PERMISSIVE_MODIFY_FOR_DN))
+ {
+ op.addRequestControl(new LDAPControl(OID_PERMISSIVE_MODIFY_CONTROL));
+ }
+ }
+
+ csn = OperationContext.getCSN(op);
+ op.run();
+
+ ResultCode result = op.getResultCode();
+
+ if (result != ResultCode.SUCCESS)
+ {
+ if (result == ResultCode.NO_OPERATION)
+ {
+ // Pre-operation conflict resolution detected that the operation
+ // was a no-op. For example, an add which has already been
+ // replayed, or a modify DN operation on an entry which has been
+ // renamed by a more recent modify DN.
+ // The change is in the data: push it to the serverState.
+ replayDone = true;
+ recordChangeResolved(csn);
+ }
+ else if (result == ResultCode.BUSY)
+ {
+ /*
+ * We probably could not get a lock (OPENDJ-885). Give the server
+ * another chance to process this operation immediately.
+ */
+ Thread.yield();
+ continue;
+ }
+ else if (isServerFailure(result, serverErrorResultCode))
+ {
+ /*
+ * It can happen when a rebuild is performed or the backend is
+ * offline (OPENDJ-49), or when the storage failed to serve the
+ * operation. Give the server another chance to process this
+ * operation after some time.
+ */
+ Thread.sleep(50);
+ continue;
+ }
+ else
+ {
+ ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO;
+ if (op instanceof ModifyOperation)
+ {
+ ModifyOperation castOp = (ModifyOperation) op;
+ dependency = remotePendingChanges.checkDependencies(castOp);
+ ModifyMsg modifyMsg = (ModifyMsg) msg;
+ resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg);
+ }
+ else if (op instanceof DeleteOperation)
+ {
+ DeleteOperation castOp = (DeleteOperation) op;
+ dependency = remotePendingChanges.checkDependencies(castOp);
+ resolution = dependency ? resolution : solveNamingConflict(castOp, msg);
+ }
+ else if (op instanceof AddOperation)
+ {
+ AddOperation castOp = (AddOperation) op;
+ AddMsg addMsg = (AddMsg) msg;
+ dependency = remotePendingChanges.checkDependencies(castOp);
+ resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg);
+ }
+ else if (op instanceof ModifyDNOperation)
+ {
+ ModifyDNOperation castOp = (ModifyDNOperation) op;
+ ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg;
+ dependency = remotePendingChanges.checkDependencies(modifyDNMsg);
+ resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg);
+ }
+ // else: unknown type of operation ?! there is nothing to replay
+
+ if (!dependency)
+ {
+ switch (resolution)
+ {
+ case NOTHING_TO_DO:
+ // the update became a dummy update and the result
+ // of the conflict resolution phase is to do nothing.
+ // however we still need to push this change to the serverState
+ replayDone = true;
+ recordChangeResolved(csn);
+ break;
+
+ case FAILED:
+ if (serverErrorResultCode.equals(result))
+ {
+ /*
+ * The result code is the one this server puts on an internal error and is
+ * one conflict resolution knows how to solve, so the change was left to it
+ * rather than treated as a failure of the server: it had its chance and
+ * could not solve it, so the storage failing is what is left. Give it the
+ * in-place attempts an UNAVAILABLE gets - a storage busy for a moment must
+ * not cost a session restart - and leave the change out of the ServerState
+ * once they are spent, which the failure of the server below the loop
+ * reports and acts on, reading the result of the attempt which spent the
+ * last of them. A change which is not in the data must not advance the
+ * ServerState (issue #889).
+ */
+ Thread.sleep(50);
+ break;
+ }
+ /*
+ * The operation did not fail on a naming conflict and not on the server
+ * either: the change can not be applied on this replica. Skip it so that the
+ * replica keeps replaying the changes which follow, but report the error in
+ * the ack and tell the administrator that the data now diverge.
+ */
+ final LocalizableMessage errorMsg = ERR_ERROR_REPLAYING_OPERATION.get(
+ op, csn, result, op.getErrorMessage());
+ logger.error(errorMsg);
+ replayErrorMsg = errorMsg.toString();
+ replayDone = true;
+ skipUnreplayableChange(csn, errorMsg);
+ break;
+
+ default:
+ /*
+ * Try replaying the change again: the next attempt creates an operation
+ * reflecting the new state of the UpdateMsg after conflict resolution
+ * modified it, and dependencies might have been replayed by now.
+ */
+ break;
+ }
+ }
+ }
+ }
+ else
+ {
+ replayDone = true;
+ }
+ }
+ }
+ finally
+ {
+ replayReadLock.unlock();
+ }
+ }
+ if (goingDown)
{
/*
* Either this replay thread or this domain is going away, or the domain is
@@ -2500,168 +2798,6 @@
replayDone = true;
break;
}
- if (!firstAttempt)
- {
- /*
- * Every attempt runs an operation of its own. An Operation which already ran
- * carries the request controls and the access log items of that run, so
- * re-running the same one stacks one ManageDsaIT control - and one access log
- * record - per attempt. It also picks up the new state of the UpdateMsg when
- * conflict resolution rewrote it.
- * Note: When msg is a DeleteMsg, the DeleteOperation is properly created
- * with subtreeDelete request control when needed.
- */
- nextOp = msg.createOperation(conn);
- }
- firstAttempt = false;
-
- // Try replay the operation
- op = nextOp;
- op.setInternalOperation(true);
- op.setSynchronizationOperation(true);
-
- // Always add the ManageDSAIT control so that updates to referrals
- // are processed locally.
- op.addRequestControl(new LDAPControl(OID_MANAGE_DSAIT_CONTROL));
-
- // Warning: specific processing ahead. See OPENDJ-2792
- if (op instanceof ModifyOperation)
- {
- ModifyOperation modifyOperation = (ModifyOperation) op;
- if (modifyOperation.getEntryDN().equals(SET_PERMISSIVE_MODIFY_FOR_DN))
- {
- op.addRequestControl(new LDAPControl(OID_PERMISSIVE_MODIFY_CONTROL));
- }
- }
-
- csn = OperationContext.getCSN(op);
- op.run();
-
- ResultCode result = op.getResultCode();
-
- if (result != ResultCode.SUCCESS)
- {
- if (result == ResultCode.NO_OPERATION)
- {
- // Pre-operation conflict resolution detected that the operation
- // was a no-op. For example, an add which has already been
- // replayed, or a modify DN operation on an entry which has been
- // renamed by a more recent modify DN.
- // The change is in the data: push it to the serverState.
- replayDone = true;
- recordChangeResolved(csn);
- }
- else if (result == ResultCode.BUSY)
- {
- /*
- * We probably could not get a lock (OPENDJ-885). Give the server
- * another chance to process this operation immediately.
- */
- Thread.yield();
- continue;
- }
- else if (isServerFailure(result, serverErrorResultCode))
- {
- /*
- * It can happen when a rebuild is performed or the backend is
- * offline (OPENDJ-49), or when the storage failed to serve the
- * operation. Give the server another chance to process this
- * operation after some time.
- */
- Thread.sleep(50);
- continue;
- }
- else
- {
- ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO;
- if (op instanceof ModifyOperation)
- {
- ModifyOperation castOp = (ModifyOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- ModifyMsg modifyMsg = (ModifyMsg) msg;
- resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg);
- }
- else if (op instanceof DeleteOperation)
- {
- DeleteOperation castOp = (DeleteOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- resolution = dependency ? resolution : solveNamingConflict(castOp, msg);
- }
- else if (op instanceof AddOperation)
- {
- AddOperation castOp = (AddOperation) op;
- AddMsg addMsg = (AddMsg) msg;
- dependency = remotePendingChanges.checkDependencies(castOp);
- resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg);
- }
- else if (op instanceof ModifyDNOperation)
- {
- ModifyDNOperation castOp = (ModifyDNOperation) op;
- ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg;
- dependency = remotePendingChanges.checkDependencies(modifyDNMsg);
- resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg);
- }
- // else: unknown type of operation ?! there is nothing to replay
-
- if (!dependency)
- {
- switch (resolution)
- {
- case NOTHING_TO_DO:
- // the update became a dummy update and the result
- // of the conflict resolution phase is to do nothing.
- // however we still need to push this change to the serverState
- replayDone = true;
- recordChangeResolved(csn);
- break;
-
- case FAILED:
- if (serverErrorResultCode.equals(result))
- {
- /*
- * The result code is the one this server puts on an internal error and is
- * one conflict resolution knows how to solve, so the change was left to it
- * rather than treated as a failure of the server: it had its chance and
- * could not solve it, so the storage failing is what is left. Give it the
- * in-place attempts an UNAVAILABLE gets - a storage busy for a moment must
- * not cost a session restart - and leave the change out of the ServerState
- * once they are spent, which the failure of the server below the loop
- * reports and acts on, reading the result of the attempt which spent the
- * last of them. A change which is not in the data must not advance the
- * ServerState (issue #889).
- */
- Thread.sleep(50);
- break;
- }
- /*
- * The operation did not fail on a naming conflict and not on the server
- * either: the change can not be applied on this replica. Skip it so that the
- * replica keeps replaying the changes which follow, but report the error in
- * the ack and tell the administrator that the data now diverge.
- */
- final LocalizableMessage errorMsg = ERR_ERROR_REPLAYING_OPERATION.get(
- op, csn, result, op.getErrorMessage());
- logger.error(errorMsg);
- replayErrorMsg = errorMsg.toString();
- replayDone = true;
- skipUnreplayableChange(csn, errorMsg);
- break;
-
- default:
- /*
- * Try replaying the change again: the next attempt creates an operation
- * reflecting the new state of the UpdateMsg after conflict resolution
- * modified it, and dependencies might have been replayed by now.
- */
- break;
- }
- }
- }
- }
- else
- {
- replayDone = true;
- }
}
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
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 03f5309..c9d7781 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/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.
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/types/LockManager.java b/opendj-server-legacy/src/main/java/org/opends/server/types/LockManager.java
index 3632615..1d641f1 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/types/LockManager.java
+++ b/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;
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index a7500d9..a66b592 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/plugins/PausePreParsePlugin.java b/opendj-server-legacy/src/test/java/org/opends/server/plugins/PausePreParsePlugin.java
new file mode 100644
index 0000000..77b38aa
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/plugins/PausePreParsePlugin.java
@@ -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();
+ }
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
index 0571660..113d380 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/FakeReplicationDomain.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/FakeReplicationDomain.java
index 90589a2..f116f66 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/FakeReplicationDomain.java
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
index 662e1a5..f9b1b1f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
+++ b/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)
diff --git a/opendj-server-legacy/tests/unit-tests-testng/resource/config-changes.ldif b/opendj-server-legacy/tests/unit-tests-testng/resource/config-changes.ldif
index 4f95eb3..8926a95 100644
--- a/opendj-server-legacy/tests/unit-tests-testng/resource/config-changes.ldif
+++ b/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
--
Gitblit v1.10.0