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 2653da06fc..3d666305c4 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
@@ -34,6 +34,7 @@
import java.io.OutputStream;
import java.io.StringReader;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
@@ -51,10 +52,12 @@
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.DataFormatException;
@@ -79,6 +82,7 @@
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.forgerock.opendj.server.config.server.ExternalChangelogDomainCfg;
import org.forgerock.opendj.server.config.server.ReplicationDomainCfg;
+import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.api.AlertGenerator;
import org.opends.server.api.DirectoryThread;
import org.opends.server.api.LocalBackend;
@@ -310,6 +314,54 @@ && getBackend().getBackendID().equals(backend.getBackendID())) {
new AtomicInteger();
/** The number of updates replayed successfully by the replication. */
private final AtomicInteger numReplayedPostOpCalled = new AtomicInteger();
+ /**
+ * How many times the replay of a change is attempted before this replica gives up on
+ * it and moves on to the changes which follow it.
+ */
+ private static final int MAX_REPLAY_ATTEMPTS = 3;
+ /**
+ * How long the session is left down before the change is asked for again, multiplied
+ * by the number of attempts already made: a backend which keeps failing must not be
+ * hammered with a session restart per failed change.
+ */
+ private static final long REPLAY_RETRY_DELAY_IN_MS = 1000;
+ /**
+ * How many changes the replay failures are remembered for. Only the changes which are
+ * failing are listed here, so the bound is never reached in practice: it is there so
+ * that a replica failing every change it is sent can not grow the map without end.
+ */
+ private static final int MAX_FAILED_REPLAY_ATTEMPTS_TRACKED = 1000;
+ /**
+ * How long the alert telling that this replica diverges is not sent again. A single
+ * cause - a schema which does not match, a backend which is gone - makes every change
+ * in flight unreplayable, and one alert per change would be a storm.
+ */
+ private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000;
+ /** The number of updates this replica gave up replaying. */
+ private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger();
+ /** Set while a replay thread is restarting the session after a failed replay. */
+ private final AtomicBoolean replayFailureRecovery = new AtomicBoolean();
+ /**
+ * How many times the replay of a change failed, keyed on the CSN of the change.
+ *
+ * The count has to be kept per change: a backend which is failing fails every change
+ * in flight, and a single slot would be reset by each of them in turn, so the give up
+ * after {@link #MAX_REPLAY_ATTEMPTS} would never be reached. Entries are dropped as
+ * soon as the change is replayed or given up on, and the oldest ones are evicted if
+ * more than {@link #MAX_FAILED_REPLAY_ATTEMPTS_TRACKED} changes are failing at once.
+ */
+ private final ConcurrentSkipListMap failedReplayAttempts = new ConcurrentSkipListMap<>();
+ /** When the alert about a change this replica gave up on was last sent. */
+ private final AtomicLong lastUnreplayedChangeAlertTime = new AtomicLong();
+ /**
+ * The result codes conflict resolution knows how to solve. The result code the server
+ * puts on an internal error is configurable and is not validated as a result code, so
+ * it could be set to one of these: it must never take a change away from
+ * {@code solveNamingConflict()}, which is the only thing which can solve them.
+ */
+ private static final Set CONFLICT_RESULT_CODES = new HashSet<>(Arrays.asList(
+ ResultCode.NO_SUCH_OBJECT, ResultCode.ENTRY_ALREADY_EXISTS,
+ ResultCode.NOT_ALLOWED_ON_RDN, ResultCode.NOT_ALLOWED_ON_NONLEAF));
private final PersistentServerState state;
private volatile boolean generationIdSavedStatus;
@@ -2001,6 +2053,8 @@ void synchronize(PostOperationOperation op)
if (op.isSynchronizationOperation())
{ // Replaying a sync operation
numReplayedPostOpCalled.incrementAndGet();
+ // The change made it, the failures it went through are history.
+ failedReplayAttempts.remove(curCSN);
try
{
remotePendingChanges.commit(curCSN);
@@ -2268,11 +2322,15 @@ public void shutdown()
/**
* Marks the specified message as the one currently processed by a replay thread.
+ *
* @param msg the message being processed
+ * @return {@code false} if the change is not pending anymore, which happens when the
+ * session was restarted after a failed replay while this message was waiting
+ * in the replay queue: it is sent again, so this copy must not be replayed
*/
- void markInProgress(LDAPUpdateMsg msg)
+ boolean markInProgress(LDAPUpdateMsg msg)
{
- remotePendingChanges.markInProgress(msg);
+ return remotePendingChanges.markInProgress(msg);
}
/**
@@ -2280,10 +2338,10 @@ void markInProgress(LDAPUpdateMsg msg)
*
* @param msg
* The UpdateMsg to be replayed.
- * @param shutdown
- * whether the server initiated shutdown
+ * @param replayThreadShutdown
+ * whether the replay thread was asked to stop
*/
- void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown)
+ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
{
// Try replay the operation, then flush (replaying) any pending operation
// whose dependency has been replayed until no more left.
@@ -2291,6 +2349,7 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown)
{
Operation op = null; // the last operation on which replay was attempted
boolean dependency = false;
+ boolean replayFailed = false;
String replayErrorMsg = null;
CSN csn = null;
try
@@ -2304,11 +2363,11 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown)
dependency = remotePendingChanges.checkDependencies(op, msg);
boolean replayDone = false;
int retryCount = 10;
- while (!dependency && !replayDone && retryCount-- > 0)
+ while (!dependency && !replayDone && !replayFailed && retryCount-- > 0)
{
- if (shutdown.get())
+ if (replayThreadShutdown.get() || shutdown.get())
{
- // shutdown initiated, let's leave
+ // Either this replay thread or this domain is going away, let's leave.
return;
}
// Try replay the operation
@@ -2343,7 +2402,9 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown)
// 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;
+ updateError(csn);
}
else if (result == ResultCode.BUSY)
{
@@ -2354,64 +2415,87 @@ else if (result == ResultCode.BUSY)
Thread.yield();
continue;
}
- else if (result == ResultCode.UNAVAILABLE)
+ else if (isServerFailure(result))
{
/*
* It can happen when a rebuild is performed or the backend is
- * offline (OPENDJ-49). Give the server another chance to process
- * this operation after some time.
+ * 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 if (op instanceof ModifyOperation)
- {
- ModifyOperation castOp = (ModifyOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- ModifyMsg modifyMsg = (ModifyMsg) msg;
- replayDone = !dependency && solveNamingConflict(castOp, modifyMsg);
- }
- else if (op instanceof DeleteOperation)
- {
- DeleteOperation castOp = (DeleteOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- replayDone = !dependency && solveNamingConflict(castOp, msg);
- }
- else if (op instanceof AddOperation)
- {
- AddOperation castOp = (AddOperation) op;
- AddMsg addMsg = (AddMsg) msg;
- dependency = remotePendingChanges.checkDependencies(castOp);
- replayDone = !dependency && solveNamingConflict(castOp, addMsg);
- }
- else if (op instanceof ModifyDNOperation)
- {
- ModifyDNOperation castOp = (ModifyDNOperation) op;
- ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg;
- dependency = remotePendingChanges.checkDependencies(modifyDNMsg);
- replayDone = !dependency && solveNamingConflict(castOp, modifyDNMsg);
- }
- else
- {
- replayDone = true; // unknown type of operation ?!
- }
-
- if (replayDone)
- {
- // 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
- updateError(csn);
- }
else
{
- /*
- * Create a new operation reflecting the new state of the UpdateMsg after conflict resolution
- * modified it and try replaying it again. Dependencies might have been replayed by now.
- * Note: When msg is a DeleteMsg, the DeleteOperation is properly
- * created with subtreeDelete request control when needed.
- */
- nextOp = msg.createOperation(conn);
+ 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;
+ updateError(csn);
+ break;
+
+ case FAILED:
+ /*
+ * 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:
+ /*
+ * Create a new operation reflecting the new state of the UpdateMsg after conflict resolution
+ * modified it and try replaying it again. Dependencies might have been replayed by now.
+ * Note: When msg is a DeleteMsg, the DeleteOperation is properly
+ * created with subtreeDelete request control when needed.
+ */
+ nextOp = msg.createOperation(conn);
+ break;
+ }
+ }
}
}
else
@@ -2420,17 +2504,34 @@ else if (op instanceof ModifyDNOperation)
}
}
- if (!replayDone && !dependency)
+ if (!replayDone && !replayFailed && !dependency)
{
- // Continue with the next change but the servers could now become
- // inconsistent.
- // Let the repair tool know about this.
- final LocalizableMessage message = ERR_LOOP_REPLAYING_OPERATION.get(
- op, op.getErrorMessage());
- logger.error(message);
- numUnresolvedNamingConflicts.incrementAndGet();
- replayErrorMsg = message.toString();
- updateError(csn);
+ if (isServerFailure(op.getResultCode()))
+ {
+ /*
+ * The server kept failing to apply the change, so the change is not in the data.
+ * Leave it out of the ServerState, otherwise the replication server would never
+ * send it again and this replica would silently diverge while reporting itself
+ * up to date.
+ */
+ final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get(
+ op, csn, op.getResultCode(), op.getErrorMessage());
+ logger.error(message);
+ replayErrorMsg = message.toString();
+ replayFailed = true;
+ }
+ else
+ {
+ // Conflict resolution kept rewriting an operation which kept failing.
+ // Continue with the next change but the servers could now become inconsistent.
+ // Let the repair tool know about this.
+ final LocalizableMessage message = ERR_LOOP_REPLAYING_OPERATION.get(
+ op, op.getErrorMessage());
+ logger.error(message);
+ numUnresolvedNamingConflicts.incrementAndGet();
+ replayErrorMsg = message.toString();
+ skipUnreplayableChange(csn, message);
+ }
}
} catch (DecodeException | LDAPException | DataFormatException e)
{
@@ -2440,9 +2541,8 @@ else if (op instanceof ModifyDNOperation)
if (csn != null)
{
/*
- * An Exception happened during the replay process.
- * Continue with the next change but the servers will now start
- * to be inconsistent.
+ * An Exception happened during the replay process: the change is not in the
+ * data, so it must not be recorded as replayed.
* Let the repair tool know about this.
*/
LocalizableMessage message =
@@ -2450,7 +2550,7 @@ else if (op instanceof ModifyDNOperation)
stackTraceToSingleLineString(e), op);
logger.error(message);
replayErrorMsg = message.toString();
- updateError(csn);
+ replayFailed = true;
} else
{
replayErrorMsg = logDecodingOperationError(msg, e);
@@ -2463,6 +2563,13 @@ else if (op instanceof ModifyDNOperation)
}
}
+ if (replayFailed && recoverFromReplayFailure(csn, replayThreadShutdown))
+ {
+ // The ack has been published and the change, still owned by the replication
+ // server, is being delivered again: there is nothing left to replay here.
+ return;
+ }
+
// Now replay any pending update that had a dependency and whose
// dependency has been replayed, do that until no more updates of that
// type left...
@@ -2485,12 +2592,16 @@ private String logDecodingOperationError(LDAPUpdateMsg msg, Exception e)
* called when error or Exceptions happen during the operation replay.
*
* @param csn the CSN of the operation with error.
+ * @return {@code false} if the change was not listed in the pending changes anymore,
+ * so that it has not been recorded as replayed: the replication server sends
+ * it again.
*/
- private void updateError(CSN csn)
+ private boolean updateError(CSN csn)
{
try
{
remotePendingChanges.commit(csn);
+ return true;
}
catch (NoSuchElementException e)
{
@@ -2502,6 +2613,201 @@ private void updateError(CSN csn)
"LDAPReplicationDomain.updateError: Unable to find remote "
+ "pending change for CSN %s", csn);
}
+ return false;
+ }
+ }
+
+ /**
+ * Returns whether the provided result code reports a failure of this server rather
+ * than a change which can not be applied: the backend being offline or rebuilt
+ * (OPENDJ-49), or the storage failing to serve the operation.
+ *
+ * @param result the result code of a replayed operation
+ * @return {@code true} if the operation failed on the server itself
+ */
+ private boolean isServerFailure(ResultCode result)
+ {
+ if (ResultCode.UNAVAILABLE.equals(result))
+ {
+ return true;
+ }
+ return result.equals(getServerContext().getCoreConfigManager().getServerErrorResultCode())
+ && !CONFLICT_RESULT_CODES.contains(result);
+ }
+
+ /**
+ * Records a change which could not be replayed as replayed anyway, so that this replica
+ * keeps replaying the changes which follow it, and warns that the data now diverge.
+ *
+ * @param csn the CSN of the change which could not be replayed
+ * @param cause the message describing why it could not be replayed
+ */
+ private void skipUnreplayableChange(CSN csn, LocalizableMessage cause)
+ {
+ failedReplayAttempts.remove(csn);
+ if (updateError(csn))
+ {
+ numFailedReplayedUpdates.incrementAndGet();
+ sendUnreplayedChangeAlert(cause);
+ }
+ // Otherwise the change is not listed as pending anymore - the session was restarted
+ // while it was being replayed - so it has not been skipped: the replication server
+ // sends it again and this replica gives up on it then.
+ }
+
+ /**
+ * Tells the administrator that this replica gave up on a change and now diverges from
+ * the rest of the topology.
+ *
+ * Whatever makes a change unreplayable - a schema which does not match, a backend
+ * which is gone - makes every change in flight unreplayable too, so the alert is not
+ * sent again for {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS}: each skipped change
+ * is logged, the alert is there to have the administrator look at the log.
+ *
+ * @param cause the message describing why the change could not be replayed
+ */
+ private void sendUnreplayedChangeAlert(LocalizableMessage cause)
+ {
+ final long now = System.currentTimeMillis();
+ final long lastSent = lastUnreplayedChangeAlertTime.get();
+ if (now - lastSent >= UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS
+ && lastUnreplayedChangeAlertTime.compareAndSet(lastSent, now))
+ {
+ DirectoryServer.sendAlertNotification(
+ this, ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, cause);
+ }
+ }
+
+ /**
+ * Lets the next change this replica gives up on raise its alert straight away.
+ *
+ * Only there for the tests which check the alert: they must not be at the mercy of the
+ * alert another test raised less than
+ * {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS} ago.
+ */
+ @VisibleForTesting
+ public void resetUnreplayedChangeAlertThrottle()
+ {
+ lastUnreplayedChangeAlertTime.set(0);
+ }
+
+ /**
+ * Counts one more failed replay of the change with the provided CSN.
+ *
+ * @param csn the CSN of the change whose replay failed
+ * @return how many times the replay of that change failed in a row
+ */
+ private int countReplayFailure(CSN csn)
+ {
+ final int attempts = failedReplayAttempts.merge(csn, 1, Integer::sum);
+ while (failedReplayAttempts.size() > MAX_FAILED_REPLAY_ATTEMPTS_TRACKED)
+ {
+ // Every change in flight is failing: keep the counts of the most recent ones,
+ // the oldest ones have been given up on long ago.
+ failedReplayAttempts.pollFirstEntry();
+ }
+ return attempts;
+ }
+
+ /**
+ * Recovers from a change which could not be replayed.
+ *
+ * The change has deliberately been left out of the ServerState, so the replication
+ * server still owns it: restart the session so that it is sent again and replayed on
+ * a backend which has hopefully recovered in the meantime. Give up once its replay
+ * failed {@link #MAX_REPLAY_ATTEMPTS} times and record it as replayed, so that a
+ * change which can never be applied here does not stop this replica for good: the
+ * administrator is told that this replica has diverged and must be reinitialized.
+ *
+ * @param csn
+ * the CSN of the change which could not be replayed, {@code null} when the
+ * message could not even be decoded
+ * @param replayThreadShutdown
+ * whether the replay thread was asked to stop
+ * @return {@code true} when the caller must stop replaying because the session is
+ * being restarted or is going away, {@code false} when it may carry on with
+ * the changes which follow
+ */
+ private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean replayThreadShutdown)
+ {
+ if (csn == null)
+ {
+ // The message could not be decoded: there is nothing to ask for again.
+ return false;
+ }
+ final int attempts = countReplayFailure(csn);
+ if (attempts >= MAX_REPLAY_ATTEMPTS)
+ {
+ final LocalizableMessage message =
+ ERR_REPLAY_SKIPPING_CHANGE.get(csn, getBaseDN(), attempts);
+ logger.error(message);
+ skipUnreplayableChange(csn, message);
+ return false;
+ }
+
+ if (replayThreadShutdown.get() || shutdown.get() || disabled)
+ {
+ /*
+ * This replay thread or this whole domain is going away: there is no session to
+ * have the change sent over again. Restarting the one which is being stopped would
+ * leave a broker and a listener thread behind on a domain whose alert generator,
+ * flush thread and RSUpdater are already gone.
+ */
+ return true;
+ }
+
+ logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), attempts);
+ if (!replayFailureRecovery.compareAndSet(false, true))
+ {
+ // Another replay thread is already restarting the session: the change it could not
+ // replay is left out of the ServerState too, so it comes back over the new session.
+ return true;
+ }
+ try
+ {
+ disableService();
+ /*
+ * The uncommitted changes are about to be sent again: forget the ones still listed
+ * as pending, or processUpdate() would discard them as duplicates. The messages
+ * they came with may still be waiting in the replay queue, which is shared by every
+ * domain and which no lock lets this thread drain in reasonable time: they are
+ * dropped when a replay thread takes them out, because markInProgress() only
+ * accepts the delivery which is listed as pending.
+ */
+ remotePendingChanges.clearUncommitted();
+ // Leave the backend some time to recover rather than ask for the change straight
+ // away: a session restart is not free for the replication server either.
+ waitBeforeSessionRestart(attempts);
+ if (shutdown.get() || disabled)
+ {
+ // The domain went away while this thread was waiting.
+ return true;
+ }
+ enableService();
+ }
+ finally
+ {
+ replayFailureRecovery.set(false);
+ }
+ return true;
+ }
+
+ /**
+ * Waits for a while before the session to the replication server is started again, so
+ * that a backend which keeps failing is not asked for every change it can not apply as
+ * fast as the replication server can send them.
+ *
+ * @param attempts how many times the replay of the change failed already
+ */
+ private void waitBeforeSessionRestart(int attempts)
+ {
+ try
+ {
+ Thread.sleep(REPLAY_RETRY_DELAY_IN_MS * attempts);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
}
}
@@ -2581,14 +2887,25 @@ private DN findEntryDN(String uuid)
return null;
}
+ /** Outcome of the conflict resolution attempted after a replayed operation failed. */
+ private enum ConflictResolution
+ {
+ /** The update message was adjusted: the operation must be replayed again. */
+ REPLAY_AGAIN,
+ /** The change is already reflected in the data: there is nothing left to replay. */
+ NOTHING_TO_DO,
+ /** The operation failed for a reason which is not a naming conflict. */
+ FAILED
+ }
+
/**
* Solve a conflict detected when replaying a modify operation.
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
- * @return true if the process is completed, false if it must continue..
+ * @return the outcome of the conflict resolution
*/
- private boolean solveNamingConflict(ModifyOperation op, ModifyMsg msg)
+ private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg)
{
ResultCode result = op.getResultCode();
ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT);
@@ -2609,14 +2926,14 @@ private boolean solveNamingConflict(ModifyOperation op, ModifyMsg msg)
// replay the modify using the current dn of this entry.
msg.setDN(newDN);
numResolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else
{
// This entry does not exist anymore.
// It has probably been deleted, stop the processing of this operation
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
}
else if (result == ResultCode.NOT_ALLOWED_ON_RDN)
@@ -2631,7 +2948,7 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN)
{
// The entry does not exist anymore.
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
// The modify operation is trying to delete the value that is
@@ -2657,15 +2974,13 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN)
}
msg.setMods(mods);
numResolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else
{
- // The other type of errors can not be caused by naming conflicts.
- // Log a message for the repair tool.
- logger.error(ERR_ERROR_REPLAYING_OPERATION,
- op, ctx.getCSN(), result, op.getErrorMessage());
- return true;
+ // The other type of errors can not be caused by naming conflicts:
+ // the operation simply failed, replay() reports it.
+ return ConflictResolution.FAILED;
}
}
@@ -2674,9 +2989,9 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN)
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
- * @return true if the process is completed, false if it must continue..
+ * @return the outcome of the conflict resolution
*/
- private boolean solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg)
+ private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg)
{
ResultCode result = op.getResultCode();
DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT);
@@ -2695,14 +3010,14 @@ private boolean solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg)
* In any case, there is nothing more to do.
*/
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
else
{
// This entry has been renamed, replay the delete using its new DN.
msg.setDN(currentDN);
numResolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
}
else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF)
@@ -2722,15 +3037,13 @@ else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF)
numUnresolvedNamingConflicts.incrementAndGet();
}
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else
{
- // The other type of errors can not be caused by naming conflicts.
- // Log a message for the repair tool.
- logger.error(ERR_ERROR_REPLAYING_OPERATION,
- op, ctx.getCSN(), result, op.getErrorMessage());
- return true;
+ // The other type of errors can not be caused by naming conflicts:
+ // the operation simply failed, replay() reports it.
+ return ConflictResolution.FAILED;
}
}
@@ -2739,10 +3052,10 @@ else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF)
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
- * @return true if the process is completed, false if it must continue.
+ * @return the outcome of the conflict resolution
* @throws Exception When the operation is not valid.
*/
-private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
+private ConflictResolution solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
throws Exception
{
ResultCode result = op.getResultCode();
@@ -2788,7 +3101,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
{
markConflictEntry(op, currentDN, currentDN.parent().child(newRDN));
numUnresolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
DN newDN = newSuperior.child(newRDN);
@@ -2801,7 +3114,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
// The entry has been deleted, we can safely assume
// that the operation is completed.
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
// if the newDN and the current DN match then the operation
@@ -2810,7 +3123,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
if (newDN.equals(currentDN))
{
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
if (result == ResultCode.NO_SUCH_OBJECT
@@ -2825,7 +3138,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg)
modifyDnMsg.setDN(currentDN);
modifyDnMsg.setNewSuperior(newSuperior.toString());
numResolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
{
@@ -2841,15 +3154,13 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
modifyDnMsg.getNewRDN()));
modifyDnMsg.setNewSuperior(newSuperior.toString());
numUnresolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else
{
- // The other type of errors can not be caused by naming conflicts.
- // Log a message for the repair tool.
- logger.error(ERR_ERROR_REPLAYING_OPERATION,
- op, ctx.getCSN(), result, op.getErrorMessage());
- return true;
+ // The other type of errors can not be caused by naming conflicts:
+ // the operation simply failed, replay() reports it.
+ return ConflictResolution.FAILED;
}
}
@@ -2858,10 +3169,10 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
*
* @param op The operation that triggered the conflict detection.
* @param msg The message that triggered the conflict detection.
- * @return true if the process is completed, false if it must continue.
+ * @return the outcome of the conflict resolution
* @throws Exception When the operation is not valid.
*/
- private boolean solveNamingConflict(AddOperation op, AddMsg msg)
+ private ConflictResolution solveNamingConflict(AddOperation op, AddMsg msg)
throws Exception
{
ResultCode result = op.getResultCode();
@@ -2884,7 +3195,7 @@ private boolean solveNamingConflict(AddOperation op, AddMsg msg)
* message for the repair tool to look at this problem.
* TODO : Log the message
*/
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
DN parentDn = findEntryDN(parentUniqueId);
if (parentDn == null)
@@ -2911,7 +3222,7 @@ private boolean solveNamingConflict(AddOperation op, AddMsg msg)
msg.setDN(DN.valueOf(msg.getDN().rdn() + "," + parentDn));
numResolvedNamingConflicts.incrementAndGet();
}
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
{
@@ -2927,7 +3238,7 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
if (findEntryDN(entryUUID) != null)
{
// entry already exist : this is a replay
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
else
{
@@ -2936,16 +3247,14 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
generateConflictRDN(entryUUID, msg.getDN().toString());
msg.setDN(DN.valueOf(conflictRDN));
numUnresolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
}
else
{
- // The other type of errors can not be caused by naming conflicts.
- // log a message for the repair tool.
- logger.error(ERR_ERROR_REPLAYING_OPERATION,
- op, ctx.getCSN(), result, op.getErrorMessage());
- return true;
+ // The other type of errors can not be caused by naming conflicts:
+ // the operation simply failed, replay() reports it.
+ return ConflictResolution.FAILED;
}
}
@@ -3847,6 +4156,8 @@ public Map getAlerts()
alerts.put(ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT,
ALERT_DESCRIPTION_REPLICATION_UNRESOLVED_CONFLICT);
+ alerts.put(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE,
+ ALERT_DESCRIPTION_REPLICATION_UNREPLAYED_CHANGE);
return alerts;
}
@@ -4258,6 +4569,12 @@ public boolean processUpdate(UpdateMsg updateMsg)
* are uncommitted changes in the queue and session failover occurs
* causing a recovery of all changes since the current committed server
* state. See OPENDJ-1115.
+ *
+ * The copy which is already listed owns the change: it is the one which records
+ * it in the ServerState once it really has been replayed. Report this delivery
+ * as done - the window and the ack are per delivery - but as handled
+ * asynchronously, so that the listener does not push the CSN to the ServerState
+ * over a change which is still being replayed or is failing (issue #889).
*/
if (logger.isTraceEnabled())
{
@@ -4265,7 +4582,8 @@ public boolean processUpdate(UpdateMsg updateMsg)
"LDAPReplicationDomain.processUpdate: ignoring "
+ "duplicate change %s", msg.getCSN());
}
- return true;
+ processUpdateDone(msg, null);
+ return false;
}
// Put update message into the replay queue
@@ -4301,6 +4619,7 @@ public void addAdditionalMonitoring(MonitorData attributes)
{
attributes.add("pending-updates", pendingChanges.size());
attributes.add("replayed-updates-ok", numReplayedPostOpCalled);
+ attributes.add("replayed-updates-failed", numFailedReplayedUpdates);
attributes.add("resolved-modify-conflicts", numResolvedModifyConflicts);
attributes.add("resolved-naming-conflicts", numResolvedNamingConflicts);
attributes.add("unresolved-naming-conflicts", numUnresolvedNamingConflicts);
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
index 06d37970f0..7d05837825 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -13,6 +13,7 @@
*
* Copyright 2007-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -140,6 +141,10 @@ public int getDependentChangesSize()
/**
* Add a new LDAPUpdateMsg that was received from the replication server
* to the pendingList.
+ *
+ * A change which is already listed is left alone: the copy which is here knows
+ * whether it has been replayed and what it depends on, while the one which comes
+ * with the message the replication server sent again knows neither.
*
* @param update The LDAPUpdateMsg that was received from the replication
* server and that will be added to the pending list.
@@ -152,7 +157,7 @@ public boolean putRemoteUpdate(LDAPUpdateMsg update)
try
{
CSN csn = update.getCSN();
- return pendingChanges.put(csn, new PendingChange(csn, null, update)) == null;
+ return pendingChanges.putIfAbsent(csn, new PendingChange(csn, null, update)) == null;
}
finally
{
@@ -200,12 +205,71 @@ public void commit(CSN csn)
}
}
- public void markInProgress(LDAPUpdateMsg msg)
+ /**
+ * Forgets the changes which have not been replayed yet, without updating the
+ * ServerState.
+ *
+ * Called when the replay of a change failed and the session to the
+ * replication server is restarted: the changes that were not committed must
+ * be replayed again, and {@link #putRemoteUpdate(LDAPUpdateMsg)} would
+ * otherwise discard the ones still listed here as duplicates.
+ *
+ * The changes which have already been committed stay: they are held back by
+ * the change which failed, so the ServerState does not cover them yet and the
+ * replication server sends them again. Forgetting them would have them
+ * replayed a second time, which is exactly what the duplicate check of
+ * {@link #putRemoteUpdate(LDAPUpdateMsg)} is there to prevent (OPENDJ-1115).
+ * Keeping them is enough: the ServerState is updated over them as soon as the
+ * change which holds them back commits.
+ */
+ public void clearUncommitted()
+ {
+ pendingChangesWriteLock.lock();
+ dependentChangesLock.lock();
+ try
+ {
+ final Iterator it = pendingChanges.values().iterator();
+ while (it.hasNext())
+ {
+ final PendingChange change = it.next();
+ if (!change.isCommitted())
+ {
+ dependentChanges.remove(change);
+ activeAndDependentChanges.remove(change);
+ it.remove();
+ }
+ }
+ }
+ finally
+ {
+ dependentChangesLock.unlock();
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
+ * Marks the change of the provided message as being replayed.
+ *
+ * @param msg
+ * the message whose change is being replayed
+ * @return {@code false} if this message is not the delivery which is listed as
+ * pending, which happens when the session was restarted after a failed
+ * replay while this message was still waiting in the replay queue: the
+ * change was either forgotten or the replication server sent it again,
+ * so this copy must not be replayed.
+ */
+ public boolean markInProgress(LDAPUpdateMsg msg)
{
pendingChangesReadLock.lock();
try
{
- activeAndDependentChanges.add(pendingChanges.get(msg.getCSN()));
+ final PendingChange change = pendingChanges.get(msg.getCSN());
+ if (change == null || change.getLDAPUpdateMsg() != msg)
+ {
+ return false;
+ }
+ activeAndDependentChanges.add(change);
+ return true;
}
finally
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
index 44ea968458..4a31d0c1d4 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2008 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -98,7 +99,15 @@ public void run()
// Find replication domain for that update message and mark it as "in progress"
updateMsg = updateToreplay.getUpdateMessage();
domain = updateToreplay.getReplicationDomain();
- domain.markInProgress(updateMsg);
+ if (!domain.markInProgress(updateMsg))
+ {
+ /*
+ * The domain restarted its session after a failed replay while this
+ * message was waiting here, so it does not know about this change
+ * anymore: the replication server sends it again over the new session.
+ */
+ continue;
+ }
}
finally
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java b/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java
index ff7f3f13bb..999bef3eaf 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java
@@ -3064,6 +3064,26 @@ public final class ServerConstants
"org.opends.server.replication.UnresolvedConflict";
+ /**
+ * The description for the alert type string that will be used for the alert
+ * notification generated when multimaster replication gives up on a change
+ * it could not replay.
+ */
+ public static final String ALERT_DESCRIPTION_REPLICATION_UNREPLAYED_CHANGE =
+ "This alert type will be used to notify administrators when " +
+ "multimaster replication gives up on a change it could not replay, " +
+ "leaving this replica diverging from the rest of the topology.";
+
+
+ /**
+ * The alert type string that will be used for the alert notification
+ * generated when multimaster replication gives up on a change it could not
+ * replay.
+ */
+ public static final String ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE =
+ "org.opends.server.replication.UnreplayedChange";
+
+
/**
* The extensible indexer identifier string that will be used for a substring
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 367a21191f..fbd10a26a5 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -610,3 +610,9 @@ ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port w
last attempt to bind it
NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \
in domain "%s" from this directory server DS(%d): rejecting the request from the remote directory server DS(%d): %s
+WARN_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (attempt %d). \
+ The change has not been recorded as replayed: restarting the session to the replication server \
+ so that it is sent again
+ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s" after %d attempts. \
+ The change is being skipped: this replica now diverges from the rest of the topology and must \
+ be reinitialized
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java b/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java
index 181b6918cb..f395262923 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java
@@ -13,10 +13,13 @@
*
* Copyright 2008 Sun Microsystems, Inc.
* Portions Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.extensions;
import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.forgerock.i18n.LocalizableMessage;
@@ -42,6 +45,9 @@ public class DummyAlertHandler
/** The number of times this alert handler has been invoked. */
private static AtomicInteger alertCount = new AtomicInteger(0);
+ /** The number of times this alert handler has been invoked, per alert type. */
+ private static final ConcurrentMap alertCountByType = new ConcurrentHashMap<>();
+
/** Creates a new instance of this SMTP alert handler. */
public DummyAlertHandler()
{
@@ -86,6 +92,7 @@ public void sendAlertNotification(AlertGenerator generator, String alertType,
LocalizableMessage alertMessage)
{
alertCount.incrementAndGet();
+ alertCountByType.computeIfAbsent(alertType, type -> new AtomicInteger()).incrementAndGet();
}
/**
@@ -98,6 +105,21 @@ public static int getAlertCount()
return alertCount.get();
}
+ /**
+ * Retrieves the number of times that this alert handler has been invoked with the
+ * provided alert type.
+ *
+ * @param alertType The type of the alert notifications to count.
+ *
+ * @return The number of times that this alert handler has been given an alert
+ * notification of that type.
+ */
+ public static int getAlertCount(String alertType)
+ {
+ final AtomicInteger count = alertCountByType.get(alertType);
+ return count != null ? count.get() : 0;
+ }
+
/** {@inheritDoc} */
@Override
public boolean isConfigurationChangeAcceptable(AlertHandlerCfg configuration,
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
index 2a04473f0a..5dc3cc1eda 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2008 Sun Microsystems, Inc.
* Portions Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.plugins;
@@ -25,6 +26,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -614,11 +616,18 @@ private int shortCircuitInternal(PluginOperation operation, String section)
}
// Check for registered short circuits.
- Integer resultCode = shortCircuits.get(
- operation.getOperationType() + "/" + section.toLowerCase());
+ final String key = operation.getOperationType() + "/" + section.toLowerCase();
+ Integer resultCode = shortCircuits.get(key);
if (resultCode != null)
{
- return resultCode;
+ final int reached = shortCircuitCounts.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet();
+ final Integer maxTimes = shortCircuitLimits.get(key);
+ if (maxTimes == null || reached <= maxTimes)
+ {
+ return resultCode;
+ }
+ // The short circuit was applied as many times as it was asked for: from now on the
+ // operations are let through, which is how a transient failure is simulated.
}
// If we've gotten here, then we shouldn't short-circuit the operation
@@ -662,6 +671,39 @@ public static List createShortCircuitControlList(int resultCode, String
/** Registered short circuits for operations regardless of controls. */
private static Map shortCircuits = new ConcurrentHashMap<>();
+ /** How many times a registered short circuit was reached. */
+ private static final Map shortCircuitCounts = new ConcurrentHashMap<>();
+
+ /** How many times a registered short circuit must be applied, when it is limited. */
+ private static final Map shortCircuitLimits = new ConcurrentHashMap<>();
+
+ /**
+ * Returns how many times the short circuit registered for the given operation type and
+ * plugin point was reached. A short circuit registered for a limited number of times is
+ * counted as reached by the operations it let through once that number was used up.
+ *
+ * @param operation The type of operation the short circuit applies to.
+ * @param section The plugin point the short circuit applies to.
+ * @return the number of operations which reached the short circuit
+ */
+ public static int getShortCircuitCount(OperationType operation, String section)
+ {
+ final AtomicInteger count = shortCircuitCounts.get(operation + "/" + section.toLowerCase());
+ return count != null ? count.get() : 0;
+ }
+
+ /**
+ * Forgets how many times the short circuit registered for the given operation type and
+ * plugin point was applied.
+ *
+ * @param operation The type of operation the short circuit applies to.
+ * @param section The plugin point the short circuit applies to.
+ */
+ public static void resetShortCircuitCount(OperationType operation, String section)
+ {
+ shortCircuitCounts.remove(operation + "/" + section.toLowerCase());
+ }
+
/**
* Register a short circuit for the given operation type and plugin point.
* @param operation The type of operation the short circuit applies to.
@@ -673,6 +715,23 @@ public static void registerShortCircuit(OperationType operation, String section,
shortCircuits.put(operation + "/" + section.toLowerCase(), resultCode);
}
+ /**
+ * Register a short circuit which only applies to the given number of operations, the
+ * ones which follow being let through: this is how a transient failure is simulated.
+ *
+ * @param operation The type of operation the short circuit applies to.
+ * @param section The plugin point the short circuit applies to.
+ * @param resultCode The result code to be returned for the short circuit.
+ * @param maxTimes How many operations must be short circuited.
+ */
+ public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes)
+ {
+ final String key = operation + "/" + section.toLowerCase();
+ shortCircuitCounts.remove(key);
+ shortCircuitLimits.put(key, maxTimes);
+ shortCircuits.put(key, resultCode);
+ }
+
/**
* Deregister a short circuit for the given operation type and plugin point.
* @param operation The type of operation the short circuit applies to.
@@ -680,6 +739,8 @@ public static void registerShortCircuit(OperationType operation, String section,
*/
public static void deregisterShortCircuit(OperationType operation, String section)
{
- shortCircuits.remove(operation + "/" + section.toLowerCase());
+ final String key = operation + "/" + section.toLowerCase();
+ shortCircuits.remove(key);
+ shortCircuitLimits.remove(key);
}
}
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 d973d2d136..ca1ad016f1 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
@@ -26,6 +26,7 @@
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
import static org.opends.server.util.CollectionUtils.*;
+import static org.opends.server.util.ServerConstants.*;
import static org.testng.Assert.*;
import java.net.SocketTimeoutException;
@@ -50,6 +51,8 @@
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.plugin.LDAPReplicationDomain;
+import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.HeartbeatThread;
@@ -1371,6 +1374,277 @@ public void call() throws Exception
}
}
+ /**
+ * Test case for [Issue 889]: a change whose replay failed on the server itself must
+ * not be recorded as replayed. Recording it would advance the ServerState past the
+ * change, so the replication server would never send it again while this replica
+ * reports itself up to date.
+ */
+ @Test
+ public void failedReplayIsNotRecordedAsReplayed() throws Exception
+ {
+ testSetUp("failedReplayIsNotRecordedAsReplayed");
+ logger.error(LocalizableMessage.raw("Starting replication test : failedReplayIsNotRecordedAsReplayed"));
+
+ final int serverId = 12;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ /*
+ * Fail the replay the way a storage failure does: the backend reports it with the
+ * server-error-result-code, 80 by default. The short circuit has to be set at the
+ * pre-parse plugin point, the pre-operation ones are not invoked for synchronization
+ * operations.
+ */
+ ShortCircuitPlugin.resetShortCircuitCount(OperationType.DELETE, "PreParse");
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ /*
+ * The replication server resumes from the ServerState of this replica, so it only
+ * sends the change again as long as the state does not cover it: seeing the same
+ * change replayed more than once is what tells that it was not recorded as
+ * replayed.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse") >= 2,
+ "the change was not sent again after its replay failed");
+ }
+ });
+ assertNotNull(getEntry(tmp.getName(), 1, true), "the entry must not have been deleted");
+
+ /*
+ * The change can never be applied here, so the replica eventually gives up on it
+ * rather than stopping for good: it then warns that it has diverged.
+ */
+ TestTimer giveUpTimer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ giveUpTimer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "the replica did not give up on a change it can never replay");
+ }
+ });
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures + 1,
+ "a change which could not be replayed must be counted once, not once per attempt");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: every change which can not be replayed must be given up
+ * on, not only the one which fails on its own.
+ *
+ * A backend which is failing fails every change in flight, which is what this test
+ * reproduces with two changes. A count kept for the last failed change only is reset
+ * by each of them in turn, so the give up would never be reached and this replica
+ * would restart its session to the replication server without end.
+ */
+ @Test
+ public void everyChangeWhichCanNotBeReplayedIsGivenUpOn() throws Exception
+ {
+ testSetUp("everyChangeWhichCanNotBeReplayedIsGivenUpOn");
+ logger.error(LocalizableMessage.raw("Starting replication test : everyChangeWhichCanNotBeReplayedIsGivenUpOn"));
+
+ final int serverId = 13;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry first = TestCaseUtils.addEntry(
+ "dn: uid=user.889.1," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.1",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ Entry second = TestCaseUtils.addEntry(
+ "dn: uid=user.889.2," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.2",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String firstUuid = getEntry(first.getName(), 1, true).parseAttribute("entryuuid").asString();
+ String secondUuid = getEntry(second.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ final CSN firstCSN = gen.newCSN();
+ final CSN secondCSN = gen.newCSN();
+ broker.publish(new DeleteMsg(first.getName(), firstCSN, firstUuid));
+ broker.publish(new DeleteMsg(second.getName(), secondCSN, secondUuid));
+
+ TestTimer giveUpTimer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ giveUpTimer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(firstCSN),
+ "the replica did not give up on the first change it can never replay");
+ assertTrue(domain.getServerState().cover(secondCSN),
+ "the replica did not give up on the second change it can never replay");
+ }
+ });
+ assertNotNull(getEntry(first.getName(), 1, true), "the first entry must not have been deleted");
+ assertNotNull(getEntry(second.getName(), 1, true), "the second entry must not have been deleted");
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures + 2,
+ "both changes must be counted as failed, once each");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: a replay which fails on the server itself is retried in
+ * place, and a failure which clears while it is retried has the change applied exactly
+ * once, without the session being restarted and without the change being reported as
+ * failed.
+ */
+ @Test
+ public void transientReplayFailureIsRetriedAndTheChangeApplied() throws Exception
+ {
+ testSetUp("transientReplayFailureIsRetriedAndTheChangeApplied");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : transientReplayFailureIsRetriedAndTheChangeApplied"));
+
+ final int serverId = 14;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.3," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.3",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ /*
+ * The backend is unavailable for the first attempts, the way it is while a rebuild
+ * is performed or while it is offline (OPENDJ-49), then it serves the operation.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.UNAVAILABLE.intValue(), 3);
+ try
+ {
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ assertNull(getEntry(tmp.getName(), 30000, false),
+ "the change was not replayed once the backend served the operation again");
+ Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
+ .as("the replay must have been retried in place")
+ .isGreaterThanOrEqualTo(4);
+
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was replayed must be recorded as replayed");
+ }
+ });
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
+ "a change which was replayed after a transient failure must not count as failed");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a transient failure must not tell the administrator that this replica diverged");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
/**
* Enable or disable the receive status of a synchronization provider.
*
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java
index f59bce4cf4..b5b31b4e5f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java
@@ -50,6 +50,7 @@
import org.opends.server.core.AddOperation;
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.DirectoryServer;
+import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.SearchRequest;
import org.opends.server.replication.ReplicationTestCase;
@@ -73,6 +74,7 @@
import org.opends.server.types.Attribute;
import org.opends.server.types.Entry;
import org.opends.server.types.Operation;
+import org.opends.server.types.OperationType;
import org.opends.server.types.SearchFilter;
import org.opends.server.types.SearchResultEntry;
import org.opends.server.util.StaticUtils;
@@ -1170,22 +1172,66 @@ public void testSafeReadModeReply(byte rsGroupId) throws Exception
return;
}
+ } finally
+ {
+ endTest(testcase);
+ }
+ }
+
+ /**
+ * Tests that a DS which could not replay an update in safe read mode acks the replay
+ * error instead of reporting the update as applied (issue #889).
+ */
+ @Test
+ public void testSafeReadModeReplyWithReplayError() throws Exception
+ {
+ int TIMEOUT = 5000;
+ String testcase = "testSafeReadModeReplyWithReplayError";
+ try
+ {
+ // Create and start a RS expecting clients in safe read assured mode
+ replicationServer = new FakeReplicationServer((byte) 1, replServerPort, RS_SERVER_ID,
+ true, testcase);
+ replicationServer.start(NO_READ);
- /* Send an update with error from the RS and get the ack with error */
+ safeReadDomainCfgEntry = createAssuredDomain(AssuredMode.SAFE_READ_MODE, 0, TIMEOUT);
+ waitForConnectionToRs(testcase, replicationServer);
+
+ Entry entry = makeEntry(
+ "dn: ou=assured-sr-replay-error-entry," + SAFE_READ_DN,
+ "objectClass: top",
+ "objectClass: organizationalUnit");
+ String parentUid = getEntryUUID(DN.valueOf(SAFE_READ_DN));
+
+ /*
+ * Fail the replay the way a storage failure does: the backend reports it with the
+ * server-error-result-code, 80 by default. The short circuit has to be set at the
+ * pre-parse plugin point, the pre-operation ones are not invoked for synchronization
+ * operations.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.ADD, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ AckMsg ackMsg = replicationServer.sendAssuredAddMsg(entry, parentUid);
- // Make the RS send a not possible assured add message
+ assertNull(DirectoryServer.getEntry(entry.getName()), "the entry must not have been added");
- // TODO: make the domain return an error: use a plugin ?
- // The resolution code does not generate any error so we need to find a
- // way to have the replay not working to test this...
+ // Check that DS replied an ack reporting the replay error
+ assertFalse(ackMsg.hasTimeout());
+ assertTrue(ackMsg.hasReplayError(), "the ack must report the failed replay");
+ assertFalse(ackMsg.hasWrongStatus());
+ Assertions.assertThat(ackMsg.getFailedServers()).containsExactly(1);
- // Check that DS replied an ack with errors
-// assertFalse(ackMsg.hasTimeout());
-// assertTrue(ackMsg.hasReplayError());
-// assertFalse(ackMsg.hasWrongStatus());
-// List failedServers = ackMsg.getFailedServers();
-// assertEquals(failedServers.size(), 1);
-// assertEquals((integer)failedServers.get(0), (integer)1);
+ /*
+ * No monitoring assertion here: the domain restarts its session to have the change
+ * sent again, which takes its monitor entry away and resets the assured counters.
+ */
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.ADD, "PreParse");
+ }
} finally
{
endTest(testcase);
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
new file mode 100644
index 0000000000..f2b0ead2be
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -0,0 +1,195 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.plugin;
+
+import static org.testng.Assert.*;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.testng.annotations.Test;
+
+/**
+ * Tests the bookkeeping a replica does on the changes it received from a replication
+ * server: a change reaches the ServerState only once it really has been replayed.
+ */
+@SuppressWarnings("javadoc")
+public class RemotePendingChangesTest extends ReplicationTestCase
+{
+ private static final int SERVER_ID = 42;
+
+ @Test
+ public void committedChangeIsPushedToTheServerState() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ assertEquals(pendingChanges.getQueueSize(), 1);
+
+ pendingChanges.commit(csn);
+
+ assertTrue(state.cover(csn));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * A change which is not committed must hold back the ServerState, even when the
+ * changes which follow it have been replayed: the replication server resumes from the
+ * ServerState, so anything it covers is never sent again.
+ */
+ @Test
+ public void uncommittedChangeHoldsBackTheChangesWhichFollowIt() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN failed = generator.newCSN();
+ final CSN next = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2")));
+
+ // The replay of the first change failed, the second one went through.
+ pendingChanges.commit(next);
+
+ assertFalse(state.cover(failed), "a change which was not replayed must not be covered");
+ assertFalse(state.cover(next), "the changes which follow a failed one must not be covered either");
+ assertEquals(pendingChanges.getQueueSize(), 2);
+
+ // The first change finally made it: both are now recorded as replayed.
+ pendingChanges.commit(failed);
+
+ assertTrue(state.cover(failed));
+ assertTrue(state.cover(next));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * Restarting the session forgets the changes which were not replayed without recording
+ * them, so that the ones the replication server sends again are replayed rather than
+ * discarded as duplicates.
+ */
+ @Test
+ public void clearUncommittedForgetsTheChangesWhichWereNotReplayed() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "a duplicate must be discarded");
+
+ pendingChanges.clearUncommitted();
+
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ assertTrue(state.isEmpty(), "clearing the pending changes must not record them as replayed");
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "the change must be accepted again once the session has been restarted");
+ }
+
+ /**
+ * A change which was replayed while an older one was still failing must survive the
+ * restart of the session. The ServerState does not cover it yet, so the replication
+ * server sends it again, and replaying it a second time is exactly what the duplicate
+ * check of {@code putRemoteUpdate()} is there to prevent (OPENDJ-1115).
+ */
+ @Test
+ public void clearUncommittedKeepsTheChangesWhichWereReplayed() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN failed = generator.newCSN();
+ final CSN next = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2")));
+ // The replay of the first change failed, the second one went through and is held
+ // back by the first one.
+ pendingChanges.commit(next);
+
+ pendingChanges.clearUncommitted();
+
+ assertEquals(pendingChanges.getQueueSize(), 1);
+ assertTrue(state.isEmpty(), "clearing the pending changes must not record them as replayed");
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2")),
+ "a change which was replayed must not be replayed a second time");
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")),
+ "the change which was not replayed must be accepted again");
+
+ // The change which failed finally made it: both are now recorded as replayed.
+ pendingChanges.commit(failed);
+
+ assertTrue(state.cover(failed));
+ assertTrue(state.cover(next));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * A message which was waiting in the replay queue while the session was restarted must
+ * be reported as not pending anymore rather than replayed against a bookkeeping which
+ * does not list its change.
+ */
+ @Test
+ public void markInProgressReportsAChangeWhichIsNotPendingAnymore() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg msg = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(msg));
+ assertTrue(pendingChanges.markInProgress(msg));
+ assertEquals(pendingChanges.changesInProgressSize(), 1);
+
+ pendingChanges.clearUncommitted();
+
+ assertEquals(pendingChanges.changesInProgressSize(), 0);
+ assertFalse(pendingChanges.markInProgress(msg),
+ "a message whose change was forgotten must not be replayed");
+ }
+
+ /**
+ * The replication server sends the change again over the new session while the message
+ * of the previous delivery may still be waiting in the replay queue: only the delivery
+ * which is listed as pending is replayed, or the same change would be applied twice.
+ */
+ @Test
+ public void markInProgressRejectsThePreviousDeliveryOfAChange() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg previousDelivery = deleteMsg(csn, "uuid-1");
+ final DeleteMsg newDelivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(previousDelivery));
+ pendingChanges.clearUncommitted();
+ assertTrue(pendingChanges.putRemoteUpdate(newDelivery), "the change must be accepted again");
+
+ assertFalse(pendingChanges.markInProgress(previousDelivery),
+ "the message of the previous delivery must not be replayed");
+ assertTrue(pendingChanges.markInProgress(newDelivery));
+ }
+
+ private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception
+ {
+ return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID);
+ }
+}