From 92d88ca699cd8090a26b92cbe46789d2b848195f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 07 Sep 2026 09:30:36 +0000
Subject: [PATCH] [#889] Keep a change the replay could not apply out of the ServerState (#892)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 263 ++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 9
opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml | 10
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java | 84 +
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java | 11
opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java | 22
opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java | 20
opendj-server-legacy/pom.xml | 2
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 728 +++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java | 99 +
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 1156 ++++++++++++++++++---
opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc | 3
opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java | 68 +
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 2
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java | 476 ++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java | 201 +++
16 files changed, 2,932 insertions(+), 222 deletions(-)
diff --git a/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc b/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc
index 5adea89..ceb1133 100644
--- a/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc
+++ b/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc
@@ -2179,6 +2179,9 @@
`org.opends.server.replication.UnresolvedConflict`::
Multimaster replication cannot resolve a conflict automatically.
+`org.opends.server.replication.UnreplayedChange`::
+Multimaster replication gave up on a change it could not replay. This replica now diverges from the rest of the topology and must be reinitialized.
+
`org.opends.server.UncaughtException`::
A directory server thread has encountered an uncaught exception that caused that thread to terminate abnormally. The impact that this problem has on the server depends on which thread was impacted and the nature of the exception.
diff --git a/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml b/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml
index e4c85d6..4194e4e 100644
--- a/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml
+++ b/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml
@@ -21,7 +21,7 @@
! CCPL HEADER END
!
! Copyright 2011-2013 ForgeRock AS
- ! Portions Copyright 2024 3A Systems, LLC
+ ! Portions Copyright 2024-2026 3A Systems, LLC
!
-->
<chapter xml:id='chap-monitoring'
@@ -984,6 +984,14 @@
</listitem>
</varlistentry>
<varlistentry>
+ <term><literal>org.opends.server.replication.UnreplayedChange</literal></term>
+ <listitem>
+ <para>Multimaster replication gave up on a change it could not replay.
+ This replica now diverges from the rest of the topology and must be
+ reinitialized.</para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
<term><literal>org.opends.server.UncaughtException</literal></term>
<listitem>
<para>A directory server thread has encountered an uncaught exception that
diff --git a/opendj-server-legacy/pom.xml b/opendj-server-legacy/pom.xml
index 7a0daad..c883bc7 100644
--- a/opendj-server-legacy/pom.xml
+++ b/opendj-server-legacy/pom.xml
@@ -1273,7 +1273,7 @@
<org.opends.test.suppressOutput>true</org.opends.test.suppressOutput>
<org.opends.test.pauseOnFailure>false</org.opends.test.pauseOnFailure>
<org.opends.test.copyClassesToTestPackage>false</org.opends.test.copyClassesToTestPackage>
- <org.opends.test.timeout>600000</org.opends.test.timeout><!--15 mins-->
+ <org.opends.test.timeout>600000</org.opends.test.timeout><!--10 mins-->
<!-- Matched against the name of the test class, see org.opends.server.TestListener.onStart(). -->
<org.opends.test.trace.pattern>(org\.opends\.server\.replication\.service\..*)|(org\.opends\.server\.replication\.GenerationIdTest)|(org\.opends\.server\.types\.HostPortTest)|(org\.openidentityplatform\.opendj\.AliasTestCase)</org.opends.test.trace.pattern>
</systemPropertyVariables>
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 0cd82bb..7fa68c0 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
@@ -55,8 +55,10 @@
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;
+
+import net.jcip.annotations.GuardedBy;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizedIllegalArgumentException;
@@ -79,6 +81,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;
@@ -138,7 +141,6 @@
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.ExistingFileBehavior;
-import org.opends.server.types.LDAPException;
import org.opends.server.types.LDIFExportConfig;
import org.opends.server.types.LDIFImportConfig;
import org.opends.server.types.Modification;
@@ -310,6 +312,120 @@
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 straight away, before the session
+ * to the replication server is restarted and the change is asked for again: a lock or
+ * a storage which is busy for a moment (OPENDJ-885) is waited out here.
+ */
+ public static final int IN_PLACE_REPLAY_ATTEMPTS = 10;
+ /**
+ * How long the replay of a change is retried before this replica gives up on it and
+ * moves on to the changes which follow it.
+ * <p>
+ * The budget is a duration rather than a number of attempts because what
+ * {@link #isServerFailure(ResultCode, ResultCode)} reports is measured in minutes: a
+ * backend which is being rebuilt, imported into or restored (OPENDJ-49) serves nothing
+ * while it works, and a handful of attempts would have this replica give up on every
+ * change of a maintenance window it only had to wait out.
+ */
+ private static final long REPLAY_GIVE_UP_DELAY_IN_MS = 300000;
+ /**
+ * 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;
+ /**
+ * The longest the session is left down between two attempts. The wait runs on one of
+ * the replay threads, which are shared by every domain of this server, so it is kept
+ * short enough not to starve the domains which are healthy.
+ */
+ private static final long MAX_REPLAY_RETRY_DELAY_IN_MS = 10000;
+ /**
+ * 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();
+ /**
+ * Set when a change whose replay failed has to be delivered again, and cleared by the
+ * replay thread which restarts the session for it. A change released while the session
+ * was being restarted has to be asked for over yet another session: the delivery this
+ * one makes is turned down as a duplicate while a replay thread still owns it.
+ */
+ private final AtomicBoolean sessionRestartRequested = new AtomicBoolean();
+ /**
+ * How many times in a row the session was restarted without a change being replayed in
+ * between. The backoff is computed from this rather than from the failures of the
+ * change which happens to open the recovery: an outage fails every change in flight,
+ * and the ones which are sent for the first time would otherwise keep the wait at its
+ * shortest for as long as the outage lasts.
+ */
+ private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger();
+ /**
+ * How long the replay of a change is retried before this replica gives up on it. Only
+ * the tests, which can not wait out {@link #REPLAY_GIVE_UP_DELAY_IN_MS}, set another
+ * value.
+ */
+ private volatile long replayGiveUpDelayInMs = REPLAY_GIVE_UP_DELAY_IN_MS;
+ /**
+ * Serialises the session of this domain being stopped and started again: the replay
+ * thread which restarts it after a failed replay must not race the domain being
+ * disabled for an import or a restore, or it would bring a broker and a listener
+ * thread back up on a domain which is supposed to be down.
+ * <p>
+ * Holding it costs something, and knowingly: {@code enableService()} connects to the
+ * replication servers under this lock, so a shutdown, an import or a configuration
+ * change which arrives while a replay thread is bringing the session back waits for
+ * that connect - up to the configured connection timeout when the replication servers
+ * are unreachable, which is the same outage that failed the replay. Every one of those
+ * stops the session as its first act, so what they wait for is a session which is about
+ * to be stopped again. The wait between the stop and the start is deliberately left
+ * outside the lock, so the waiting is bounded by a connect rather than by the backoff.
+ */
+ private final Object serviceStateLock = new Object();
+ /**
+ * Bumped every time the session of this domain is stopped or started under
+ * {@link #serviceStateLock}. A replay thread which stopped the session only starts it
+ * back if this still is the session it stopped: a configuration change, or the end of
+ * an import, may have started another one while it was waiting for the backend to
+ * recover.
+ * <p>
+ * It does not count the sessions {@code changeConfig()} and {@code readAssuredConfig()}
+ * stop and start, which they do without knowing about it: they run under the lock, so a
+ * replay thread never observes one of theirs, but a session it stopped may well have
+ * been replaced by one of theirs while it was waiting. That is why the guard in
+ * {@link #restartSession(boolean)} reads {@code isListenerShuttingDown()} as well - a
+ * session started outside this counter leaves it untouched, and only the listener says
+ * that one is running.
+ */
+ @GuardedBy("serviceStateLock")
+ private long sessionGeneration;
+ /**
+ * 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.
+ */
+ private static final long UNREPLAYED_CHANGE_ALERT_NEVER_SENT = Long.MIN_VALUE / 2;
+ /** When the alert about a change this replica gave up on was last sent. */
+ private final AtomicLong lastUnreplayedChangeAlertTime =
+ new AtomicLong(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
+ /**
+ * 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<ResultCode> CONFLICT_RESULT_CODES = Collections.unmodifiableSet(
+ newHashSet(
+ ResultCode.NO_SUCH_OBJECT, ResultCode.ENTRY_ALREADY_EXISTS,
+ ResultCode.NOT_ALLOWED_ON_RDN, ResultCode.NOT_ALLOWED_ON_NONLEAF,
+ // solveNamingConflict(ModifyDNOperation) solves these two as well
+ ResultCode.UNWILLING_TO_PERFORM, ResultCode.OBJECTCLASS_VIOLATION));
private final PersistentServerState state;
private volatile boolean generationIdSavedStatus;
@@ -682,36 +798,49 @@
// Disable service if configuration changed
final boolean needRestart = needReconnection && allowReconnection;
- if (needRestart)
+ /*
+ * The session is stopped, the configuration it depends on is changed and the session
+ * is started again under the lock which the replay thread restarting the session after
+ * a failed replay holds too: a session brought up in the middle of this would be
+ * reading a fractional configuration which is half way through being changed. The
+ * pair has to be atomic, which the lock inside disableService()/enableService() does
+ * not make it.
+ */
+ synchronized (serviceStateLock)
{
- disableService();
- }
- // Set new configuration
- int newFractionalMode = newFractionalConfig.fractionalConfigToInt();
- fractionalConfig.setFractional(newFractionalMode !=
- FractionalConfig.NOT_FRACTIONAL);
- if (fractionalConfig.isFractional())
- {
- // Set new fractional configuration values
- fractionalConfig.setFractionalExclusive(
- newFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL);
- fractionalConfig.setFractionalSpecificClassesAttributes(
- newFractionalConfig.getFractionalSpecificClassesAttributes());
- fractionalConfig.setFractionalAllClassesAttributes(
- newFractionalConfig.fractionalAllClassesAttributes);
- } else
- {
- // Reset default values
- fractionalConfig.setFractionalExclusive(true);
- fractionalConfig.setFractionalSpecificClassesAttributes(
- new HashMap<String, Set<String>>());
- fractionalConfig.setFractionalAllClassesAttributes(new HashSet<String>());
- }
+ if (needRestart)
+ {
+ disableService();
+ sessionGeneration++;
+ }
+ // Set new configuration
+ int newFractionalMode = newFractionalConfig.fractionalConfigToInt();
+ fractionalConfig.setFractional(newFractionalMode !=
+ FractionalConfig.NOT_FRACTIONAL);
+ if (fractionalConfig.isFractional())
+ {
+ // Set new fractional configuration values
+ fractionalConfig.setFractionalExclusive(
+ newFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL);
+ fractionalConfig.setFractionalSpecificClassesAttributes(
+ newFractionalConfig.getFractionalSpecificClassesAttributes());
+ fractionalConfig.setFractionalAllClassesAttributes(
+ newFractionalConfig.fractionalAllClassesAttributes);
+ } else
+ {
+ // Reset default values
+ fractionalConfig.setFractionalExclusive(true);
+ fractionalConfig.setFractionalSpecificClassesAttributes(
+ new HashMap<String, Set<String>>());
+ fractionalConfig.setFractionalAllClassesAttributes(new HashSet<String>());
+ }
- // Reconnect if required
- if (needRestart)
- {
- enableService();
+ // Reconnect if required
+ if (needRestart)
+ {
+ enableService();
+ sessionGeneration++;
+ }
}
}
@@ -2010,6 +2139,7 @@
logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN);
return;
}
+ resetSessionRestartBackoff();
}
else
{
@@ -2249,8 +2379,15 @@
.deregisterLocalBackendInitializationListener(this);
DirectoryServer.deregisterShutdownListener(this);
- // stop the ReplicationDomain
- disableService();
+ // stop the ReplicationDomain, under the lock which the replay thread restarting the
+ // session after a failed replay holds: it must not bring a broker and a listener
+ // thread back up on a domain whose alert generator, flush thread and RSUpdater have
+ // just been taken away.
+ synchronized (serviceStateLock)
+ {
+ disableService();
+ sessionGeneration++;
+ }
}
// wait for completion of the ServerStateFlush thread.
@@ -2268,11 +2405,33 @@
/**
* 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);
+ if (remotePendingChanges.markInProgress(msg))
+ {
+ return true;
+ }
+ /*
+ * This delivery is not replayed, but it was taken off the replay queue all the same:
+ * count it as processed, or replication-processed-updates would drift away from what
+ * the session delivered - that attribute counts the deliveries this replica took off
+ * the session, not the ones which reached the replay. No ack is owed for it either
+ * way, though not always for the same reason. When the change was taken over, this
+ * copy came over a session which has been torn down since and the delivery which took
+ * over from it carries the ack. When a disabled domain forgot the change, no delivery
+ * takes over and none is acked: the session that one came over is gone as well, so an
+ * ack would reach nobody, and a server waiting on an assured write times it out the
+ * way it does for every change in flight when a domain is taken out of the topology
+ * for an import or a restore.
+ */
+ incProcessedUpdates();
+ return false;
}
/**
@@ -2280,10 +2439,10 @@
*
* @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 +2450,8 @@
{
Operation op = null; // the last operation on which replay was attempted
boolean dependency = false;
+ boolean replayFailed = false;
+ boolean replayAbandoned = false;
String replayErrorMsg = null;
CSN csn = null;
try
@@ -2301,16 +2462,62 @@
// "op" is already initialized to the next Operation because of the
// error handling paths.
Operation nextOp = op = msg.createOperation(conn);
+ /*
+ * The code this server puts on an internal error is a configuration knob: it is
+ * read once here so that every attempt of this delivery, and the verdict which
+ * follows them, are judged against the same one. Read inside the try - the ack of
+ * a delivery is published in the finally below, whatever the delivery ran into -
+ * and once the operation is built, so that a failure of the read is a failure of
+ * an attempt rather than a message no operation could be built from.
+ */
+ final ResultCode serverErrorResultCode =
+ getServerContext().getCoreConfigManager().getServerErrorResultCode();
dependency = remotePendingChanges.checkDependencies(op, msg);
boolean replayDone = false;
- int retryCount = 10;
+ boolean firstAttempt = true;
+ int retryCount = IN_PLACE_REPLAY_ATTEMPTS;
while (!dependency && !replayDone && retryCount-- > 0)
{
- if (shutdown.get())
+ if (replayThreadShutdown.get() || shutdown.get() || disabled)
{
- // shutdown initiated, let's leave
- return;
+ /*
+ * Either this replay thread or this domain is going away, or the domain is
+ * being imported into or restored, so let's leave. The change was never
+ * applied, so the ack says so - an assured write must not be told that a
+ * change this replica is asking for again is in the data here - and the change
+ * is given back to the replication server rather than left listed as being
+ * replayed by a thread which is gone. Handing it back stops and starts the
+ * session, so it waits until the ack has been published on the session this
+ * delivery came over.
+ *
+ * A disabled domain saved its ServerState, cleared it from memory and forgot
+ * its pending changes: a thread which kept applying changes into the backend
+ * being imported into would have every one of its commits fail on a map which
+ * is empty, one ERR_OPERATION_NOT_FOUND_IN_PENDING per change in flight, and
+ * would be writing into a backend the import owns. abandonReplay() knows there
+ * is nothing left to hand back in that case.
+ */
+ replayErrorMsg =
+ NOTE_REPLAY_ABANDONED_CHANGE.get(msg.getCSN(), getBaseDN()).toString();
+ replayAbandoned = true;
+ replayDone = true;
+ break;
}
+ if (!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);
@@ -2343,7 +2550,9 @@
// 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)
{
@@ -2354,64 +2563,102 @@
Thread.yield();
continue;
}
- else if (result == ResultCode.UNAVAILABLE)
+ else if (isServerFailure(result, serverErrorResultCode))
{
/*
* 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 ?!
- }
+ 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 (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);
+ 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
@@ -2422,47 +2669,125 @@
if (!replayDone && !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);
+ final ResultCode lastResult = op.getResultCode();
+ /*
+ * BUSY is a lock which could not be taken (OPENDJ-885): the in-place attempts
+ * only yield to the thread which holds it, so a lock held for a while burns
+ * every one of them in no time. It is as transient as a storage which failed,
+ * and the change is just as absent from the data. So is a change whose last
+ * attempt came back with the configured server-error-result-code, whether or
+ * not conflict resolution owns that code: it had its chance, and what is left
+ * is the storage failing to serve the operation. The result of that attempt is
+ * what decides, so that this branch reports the failure it is acting on: an
+ * attempt which ended on something conflict resolution kept rewriting is the
+ * loop below, however the attempts before it ended.
+ */
+ if (isServerFailure(lastResult, serverErrorResultCode)
+ || ResultCode.BUSY.equals(lastResult)
+ || serverErrorResultCode.equals(lastResult))
+ {
+ /*
+ * 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, lastResult, 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)
- {
- replayErrorMsg = logDecodingOperationError(msg, e);
} catch (Exception e)
{
- if (csn != null)
+ if (op == null)
{
/*
- * An Exception happened during the replay process.
- * Continue with the next change but the servers will now start
- * to be inconsistent.
- * Let the repair tool know about this.
+ * No operation could be built from this message: createOperation() threw,
+ * whether it said so with a decoding exception or with an unchecked one. There
+ * is nothing to retry and no delivery which would build one any better.
+ *
+ * The decoding exceptions are caught here rather than in a catch of their
+ * own because such a catch would span the whole replay: addConflict(), which
+ * solveNamingConflict() calls once the operation has run, declares one, and a
+ * change whose operation ran must never be given up on where it failed.
*/
- LocalizableMessage message =
- ERR_EXCEPTION_REPLAYING_OPERATION.get(
- stackTraceToSingleLineString(e), op);
+ replayErrorMsg = giveUpOnUndecodableChange(msg, e);
+ }
+ else
+ {
+ /*
+ * 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.
+ *
+ * The operation was built, so whatever failed is a failure of this attempt
+ * rather than a verdict on every delivery of the change - including a failure
+ * before the CSN of the operation was read, such as the entry DN of a
+ * ModifyMsg which does not parse leaving getEntryDN() null. Giving up on it
+ * where it is reported would record a change which never reached the backend
+ * in the ServerState, which is issue #889 by another route; it is left out of
+ * the ServerState and asked for again instead, and the give-up budget bounds
+ * how long this replica keeps asking.
+ */
+ final LocalizableMessage message =
+ ERR_EXCEPTION_REPLAYING_OPERATION.get(op, stackTraceToSingleLineString(e));
logger.error(message);
replayErrorMsg = message.toString();
- updateError(csn);
- } else
- {
- replayErrorMsg = logDecodingOperationError(msg, e);
+ replayFailed = true;
}
} finally
{
if (!dependency)
{
+ /*
+ * The ack is per delivery, and it says what this delivery did: a change which
+ * failed is not in the data now, whether or not the delivery which follows
+ * manages to apply it. Holding the ack back until the change is resolved would
+ * not tell the truth any better - the session it came over is about to be torn
+ * down, so nothing would reach the server which is waiting for it, and an
+ * assured write would wait out its timeout rather than be told what happened.
+ */
processUpdateDone(msg, replayErrorMsg);
}
}
+ if (replayAbandoned)
+ {
+ /*
+ * The ack has been published, so the change can be handed back now and the session
+ * restarted for it: this thread is on its way out either way.
+ */
+ abandonReplay(msg.getCSN());
+ return;
+ }
+
+ /*
+ * The CSN of the change is read off the message rather than off the operation: the
+ * two are the same - the operation carries the CSN of the message it was built from
+ * - but a failure which happened before the operation was built, or before its CSN
+ * was read, has a change to ask for again all the same.
+ */
+ if (replayFailed && recoverFromReplayFailure(msg.getCSN(), 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...
@@ -2470,11 +2795,37 @@
} while (msg != null);
}
- private String logDecodingOperationError(LDAPUpdateMsg msg, Exception e)
+ /**
+ * Reports a message this replica can not turn into an operation, and gives up on the
+ * change it carries.
+ * <p>
+ * There is no operation to retry and no delivery which would decode any better, so the
+ * change is skipped rather than left out of the ServerState: a change which stays
+ * listed and uncommitted is the barrier which holds this domain's ServerState - and
+ * every change which follows it, from every master - back for good, since nothing asks
+ * for it again and the delivery which would is turned down while a replay thread still
+ * owns it. Skipping it says out loud what a wedged domain would only have implied: this
+ * replica has diverged and must be reinitialized.
+ * <p>
+ * Only a message which no operation could be built from comes here, and it is the
+ * {@code op == null} of its single caller which says so rather than the type of the
+ * exception: a decoding exception is declared past the point where the operation ran
+ * as well - by addConflict() - so a catch which read the type would give up on a
+ * change the backend may well have applied. A failure of the replay of an operation
+ * which was built, whenever it happens, keeps its change out of the ServerState and
+ * has it delivered again instead: that one is a failure of an attempt, not of every
+ * delivery of the change.
+ *
+ * @param msg the message which could not be decoded
+ * @param e the failure to decode it
+ * @return the error to report in the ack of this delivery
+ */
+ private String giveUpOnUndecodableChange(LDAPUpdateMsg msg, Exception e)
{
LocalizableMessage message =
ERR_EXCEPTION_DECODING_OPERATION.get(msg + " " + stackTraceToSingleLineString(e));
logger.error(message);
+ skipUnreplayableChange(msg.getCSN(), message);
return message.toString();
}
@@ -2485,12 +2836,16 @@
* 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,10 +2857,389 @@
"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
+ * @param serverErrorResultCode the result code this server puts on an internal error
+ * @return {@code true} if the operation failed on the server itself
+ */
+ private static boolean isServerFailure(ResultCode result, ResultCode serverErrorResultCode)
+ {
+ /*
+ * The result code the server puts on an internal error is configurable and is not
+ * validated as a result code, so it may well be one conflict resolution knows how to
+ * solve: such a setting must not take a change away from solveNamingConflict(), which
+ * is the only thing which can solve them. A change it could not solve either is a
+ * failure of the server all the same, which replay() acts on once conflict resolution
+ * has reported it.
+ */
+ return ResultCode.UNAVAILABLE.equals(result)
+ || (serverErrorResultCode.equals(result) && !CONFLICT_RESULT_CODES.contains(result));
+ }
+
+ /**
+ * Returns a time which only ever moves forward, in milliseconds.
+ * <p>
+ * How long a change has been failing and how long ago the last alert was sent are
+ * durations rather than dates: the wall clock stepping backwards must not have this
+ * replica retry a change for good, and stepping forwards must not have it give up on a
+ * change it had only just started to retry.
+ *
+ * @return the number of milliseconds since an arbitrary origin
+ */
+ private static long monotonicNowInMs()
+ {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
+ }
+
+ /**
+ * Records a change which conflict resolution found nothing left to do for: the change
+ * is in the data, so it is recorded as replayed and the failures it went through - and
+ * the session restarts they caused - are history.
+ *
+ * @param csn the CSN of the change
+ */
+ private void recordChangeResolved(CSN csn)
+ {
+ updateError(csn);
+ resetSessionRestartBackoff();
+ }
+
+ /**
+ * Has the change which fails next start the backoff between the session restarts over,
+ * if this replica is not failing any change anymore.
+ * <p>
+ * A change made it and nothing is failing anymore, so the backend is serving again and
+ * the session is not being restarted in a row. While something is still failing, a
+ * change which was replayed says nothing of the kind - a change which can never be
+ * applied here fails alone, among changes which replay perfectly well, and letting
+ * those reset the wait would have this domain tear its session down every second for as
+ * long as that one change takes to be given up on.
+ */
+ private void resetSessionRestartBackoff()
+ {
+ if (!remotePendingChanges.hasFailingChanges())
+ {
+ consecutiveSessionRestarts.set(0);
+ }
+ }
+
+ /**
+ * 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.
+ * <p>
+ * The backoff between the session restarts is deliberately left alone: giving up on a
+ * change is not a change being replayed, and whatever made this one unreplayable is
+ * still failing the ones which are in flight with it. Restarting it from its shortest
+ * wait would have a replica which gives up on a change now and then ask for every
+ * change of an outage as fast as the replication server can send them, which is what
+ * {@link #consecutiveSessionRestarts} is there to prevent.
+ *
+ * @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)
+ {
+ if (updateError(csn))
+ {
+ numFailedReplayedUpdates.incrementAndGet();
+ sendUnreplayedChangeAlert(cause);
+ }
+ // Otherwise the change is not listed as pending anymore - the domain was disabled
+ // 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.
+ * <p>
+ * 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 = monotonicNowInMs();
+ 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.
+ * <p>
+ * 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(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
+ }
+
+ /**
+ * Recovers from a change which could not be replayed.
+ * <p>
+ * 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 has
+ * been failing for {@link #REPLAY_GIVE_UP_DELAY_IN_MS} 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
+ * @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)
+ {
+ /*
+ * The failure is recorded, and the change given up on, while this thread still owns
+ * it: a change which is listed, uncommitted and unowned is what putRemoteUpdate()
+ * takes over, so releasing it before the decision is made would let another delivery
+ * be replayed by another thread while this one goes on to record the change as
+ * skipped.
+ */
+ final long now = monotonicNowInMs();
+ final RemotePendingChanges.ReplayFailure failure =
+ remotePendingChanges.recordReplayFailure(csn, now);
+ if (failure == null)
+ {
+ /*
+ * There is no uncommitted change left to give up on or to ask for again: the domain
+ * was disabled while this one was being replayed - its ServerState was saved and is
+ * read again from the backend when the domain is enabled back - or the change had
+ * already been recorded when this failure was reported. Carry on with the changes
+ * which were waiting for it; a domain on its way down lists none.
+ */
+ return false;
+ }
+ if (failure.getFailingForMs() >= replayGiveUpDelayInMs)
+ {
+ final LocalizableMessage message = ERR_REPLAY_SKIPPING_CHANGE.get(
+ csn, getBaseDN(), failure.getFailingForMs(), failure.getAttempts());
+ logger.error(message);
+ skipUnreplayableChange(csn, message);
+ return false;
+ }
+
+ /*
+ * The change stays listed as pending and uncommitted - it holds the ServerState back
+ * so that the replication server sends it again - but this thread does not own it
+ * anymore: the next delivery is the one which is replayed, and the copy which may
+ * still wait in the shared replay queue is dropped when a replay thread takes it out,
+ * because markInProgress() only accepts the delivery which is listed as pending.
+ */
+ remotePendingChanges.replayFailed(csn);
+
+ if (shutdown.get() || disabled)
+ {
+ /*
+ * This whole domain is going away or is being imported into: there is no session of
+ * this thread's to restart. 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(), failure.getAttempts());
+ /*
+ * This change is not owned by anyone anymore, so the session has to be restarted for
+ * the replication server to deliver it again. Ask for the restart before trying to
+ * run it: a restart which is already under way may have started before this change
+ * was released, and the delivery it asked for would then have been turned down as a
+ * duplicate of a change a replay thread still owned.
+ */
+ sessionRestartRequested.set(true);
+ /*
+ * A replay thread which is stopping - the number of them is being changed - restarts
+ * the session all the same: nothing else would ask for the change it just released,
+ * and the ServerState would stay behind it for good. It does not sit through the
+ * backoff on its way out, though: the backend is not what is going away.
+ */
+ runRequestedSessionRestarts(!replayThreadShutdown.get());
+ return true;
+ }
+
+ /**
+ * Restarts the session as long as changes which could not be replayed are waiting to be
+ * delivered again.
+ *
+ * @param wait whether to leave the backend some time to recover between two restarts
+ */
+ private void runRequestedSessionRestarts(boolean wait)
+ {
+ /*
+ * The outer loop is what makes a request which was made while this thread was giving
+ * up the recovery its own: the thread which made it found the recovery taken and left
+ * it to this one.
+ */
+ while (sessionRestartRequested.get() && replayFailureRecovery.compareAndSet(false, true))
+ {
+ try
+ {
+ while (sessionRestartRequested.getAndSet(false))
+ {
+ restartSession(wait);
+ }
+ }
+ finally
+ {
+ replayFailureRecovery.set(false);
+ }
+ }
+ }
+
+ /**
+ * Gives a change back to the replication server when this replay thread stops before it
+ * could apply it.
+ * <p>
+ * The change is not owned by anyone anymore and it was never applied, so the session is
+ * restarted for it to be delivered again: it is left out of the ServerState, and the
+ * changes which follow it are held back until it is replayed.
+ *
+ * @param csn the CSN of the change this thread was replaying
+ */
+ private void abandonReplay(CSN csn)
+ {
+ remotePendingChanges.replayFailed(csn);
+ if (shutdown.get() || disabled)
+ {
+ // The domain owns its session, and it forgets its pending changes on its way down.
+ return;
+ }
+ /*
+ * Logged here rather than where the change is abandoned: a server which is shutting
+ * down abandons every change in flight, and none of them is asked for again before it
+ * is started back - one line per change would say otherwise.
+ */
+ logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN());
+ sessionRestartRequested.set(true);
+ runRequestedSessionRestarts(false);
+ }
+
+ /**
+ * Stops the session to the replication server and starts it again, so that the changes
+ * this replica could not replay are delivered again.
+ */
+ private void restartSession(boolean wait)
+ {
+ final long stoppedSession;
+ synchronized (serviceStateLock)
+ {
+ if (shutdown.get() || disabled)
+ {
+ // The domain is going away or is being imported into: it owns its session.
+ return;
+ }
+ disableService();
+ stoppedSession = ++sessionGeneration;
+ }
+ if (wait)
+ {
+ /*
+ * 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. The wait
+ * is not held under the lock, or a domain being disabled for an import would wait
+ * it out.
+ */
+ waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet());
+ }
+ synchronized (serviceStateLock)
+ {
+ if (shutdown.get() || disabled
+ || sessionGeneration != stoppedSession || !isListenerShuttingDown())
+ {
+ /*
+ * The domain went away while this thread was waiting, or the session was stopped
+ * and started again by something else - a configuration change, the end of an
+ * import - in the meantime: the session this thread stopped is gone, so it has
+ * nothing left to start. The generation says a session was started under this
+ * lock; the listener says one is running, which is what a restart made outside it
+ * leaves behind.
+ */
+ return;
+ }
+ enableService();
+ sessionGeneration++;
+ }
+ }
+
+ /**
+ * 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 restarts how many times in a row the session was restarted already
+ */
+ private void waitBeforeSessionRestart(int restarts)
+ {
+ try
+ {
+ Thread.sleep(Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS));
+ }
+ catch (InterruptedException e)
+ {
+ /*
+ * Do not wait, but do start the session again all the same: the session is down
+ * because this thread stopped it, and leaving it down would take this domain out of
+ * the topology until the server is restarted. The interrupt is left set for whoever
+ * asked this thread to stop.
+ */
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Returns how long the replay of a change is retried before this replica gives up on
+ * it.
+ * <p>
+ * Only there for the tests, which set another value and put this one back.
+ *
+ * @return how long a change is retried, in milliseconds
+ */
+ @VisibleForTesting
+ public long getReplayGiveUpDelay()
+ {
+ return replayGiveUpDelayInMs;
+ }
+
+ /**
+ * Sets how long the replay of a change is retried before this replica gives up on it.
+ * <p>
+ * Only there for the tests, which can not wait out the {@link
+ * #REPLAY_GIVE_UP_DELAY_IN_MS} a backend under maintenance is given.
+ *
+ * @param delayInMs how long a change is retried, in milliseconds
+ */
+ @VisibleForTesting
+ public void setReplayGiveUpDelay(long delayInMs)
+ {
+ this.replayGiveUpDelayInMs = delayInMs;
+ }
+
+ /**
* Generate a new CSN and insert it in the pending list.
*
* @param operation
@@ -2581,14 +3315,25 @@
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 +3354,14 @@
// 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 +3376,7 @@
{
// 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 +3402,13 @@
}
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 +3417,9 @@
*
* @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 +3438,14 @@
* 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 +3465,13 @@
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 +3480,10 @@
*
* @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 +3529,7 @@
{
markConflictEntry(op, currentDN, currentDN.parent().child(newRDN));
numUnresolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
DN newDN = newSuperior.child(newRDN);
@@ -2801,7 +3542,7 @@
// 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 +3551,7 @@
if (newDN.equals(currentDN))
{
numResolvedNamingConflicts.incrementAndGet();
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
if (result == ResultCode.NO_SUCH_OBJECT
@@ -2825,7 +3566,7 @@
modifyDnMsg.setDN(currentDN);
modifyDnMsg.setNewSuperior(newSuperior.toString());
numResolvedNamingConflicts.incrementAndGet();
- return false;
+ return ConflictResolution.REPLAY_AGAIN;
}
else if (result == ResultCode.ENTRY_ALREADY_EXISTS)
{
@@ -2841,15 +3582,13 @@
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 +3597,10 @@
*
* @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 +3623,7 @@
* 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 +3650,7 @@
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 +3666,7 @@
if (findEntryDN(entryUUID) != null)
{
// entry already exist : this is a replay
- return true;
+ return ConflictResolution.NOTHING_TO_DO;
}
else
{
@@ -2936,16 +3675,14 @@
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;
}
}
@@ -3129,10 +3866,30 @@
*/
public void disable()
{
- state.save();
- state.clearInMemory();
- disabled = true;
- disableService(); // This will cut the session and wake up the listener
+ synchronized (serviceStateLock)
+ {
+ state.save();
+ state.clearInMemory();
+ disabled = true;
+ disableService(); // This will cut the session and wake up the listener
+ sessionGeneration++;
+ /*
+ * 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
+ * pending must not outlive it: one which stayed would be discarded as a duplicate
+ * when the replication server sends it again, and nothing would ever replay it or
+ * record it in the ServerState. The listener thread is stopped first, or a change
+ * it lists after this would be the one left behind.
+ */
+ remotePendingChanges.clear();
+ /*
+ * The recovery from a failed replay is over as well: the change it was asking for
+ * is gone with the pending changes, so a leftover request would have a replay thread
+ * stop and start the session once for a delivery which can not come.
+ */
+ sessionRestartRequested.set(false);
+ consecutiveSessionRestarts.set(0);
+ }
}
/**
@@ -3163,23 +3920,27 @@
*/
public void enable()
{
- try
+ synchronized (serviceStateLock)
{
- loadDataState();
- }
- catch (Exception e)
- {
- /* TODO should mark that replicationServer service is
- * not available, log an error and retry upon timeout
- * should we stop the modifications ?
- */
- logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e));
- return;
- }
+ try
+ {
+ loadDataState();
+ }
+ catch (Exception e)
+ {
+ /* TODO should mark that replicationServer service is
+ * not available, log an error and retry upon timeout
+ * should we stop the modifications ?
+ */
+ logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e));
+ return;
+ }
- enableService();
+ enableService();
+ sessionGeneration++;
- disabled = false;
+ disabled = false;
+ }
}
/**
@@ -3795,11 +4556,20 @@
ReplicationDomainCfg configuration)
{
this.config = configuration;
- changeConfig(configuration);
+ /*
+ * Each of these stops and starts the session when what it changes calls for it, and
+ * the configuration they change is read as the session comes up: hold the lock the
+ * replay thread restarting the session after a failed replay takes, so that none of
+ * them is interleaved with a session it did not start itself.
+ */
+ synchronized (serviceStateLock)
+ {
+ changeConfig(configuration);
- // Read assured + fractional configuration and each time reconnect if needed
- readAssuredConfig(configuration, true);
- readFractionalConfig(configuration, true);
+ // Read assured + fractional configuration and each time reconnect if needed
+ readAssuredConfig(configuration, true);
+ readFractionalConfig(configuration, true);
+ }
solveConflictFlag = isSolveConflict(configuration);
@@ -3847,6 +4617,8 @@
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;
}
@@ -4254,10 +5026,20 @@
if (!remotePendingChanges.putRemoteUpdate(msg))
{
/*
- * Already received this change so ignore it. This may happen if there
- * 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.
+ * A replay thread already owns this change, so this delivery is a duplicate:
+ * ignore it. This happens when there 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).
+ *
+ * A change whose replay failed is not owned by anyone anymore, so this is not the
+ * path it takes: putRemoteUpdate() takes this delivery over from the one which
+ * failed and has it replayed again.
*/
if (logger.isTraceEnabled())
{
@@ -4265,7 +5047,8 @@
"LDAPReplicationDomain.processUpdate: ignoring "
+ "duplicate change %s", msg.getCSN());
}
- return true;
+ processUpdateDone(msg, null);
+ return false;
}
// Put update message into the replay queue
@@ -4301,6 +5084,7 @@
{
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/PendingChange.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
index 38f0661..34f4c4f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.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.replication.plugin;
@@ -29,7 +30,29 @@
{
private final CSN csn;
private boolean committed;
- private UpdateMsg msg;
+ /**
+ * Written when the delivery which owns a remote change is taken over by the one which
+ * follows it, and read by the dependency checks without the pending changes lock.
+ */
+ private volatile UpdateMsg msg;
+ /**
+ * Whether a replay thread owns this change: it is being replayed, or it waits for the
+ * change it depends on. A remote change which no thread owns is one whose replay
+ * failed and which the replication server is expected to deliver again.
+ */
+ private boolean owned;
+ /**
+ * How many times in a row the replay of this change failed, and when the first of
+ * those failures happened - on a clock which only moves forward.
+ * <p>
+ * They live here, on the change which stays listed as the barrier holding the
+ * ServerState back, rather than in a map on the side: a bound on such a map would have
+ * a change evicted between two of its own failures and its give-up budget restarted,
+ * so a replica failing more changes than the bound would never give up on any of them
+ * (issue #889).
+ */
+ private int replayFailures;
+ private long firstReplayFailureTimeMs;
private final PluginOperation op;
/**
@@ -108,6 +131,65 @@
}
/**
+ * Returns whether a replay thread owns this change.
+ *
+ * @return {@code true} if a replay thread is replaying this change or waiting for the
+ * change it depends on
+ */
+ public boolean isOwned()
+ {
+ return owned;
+ }
+
+ /**
+ * Sets whether a replay thread owns this change.
+ *
+ * @param owned {@code true} when a replay thread takes the change over, {@code false}
+ * when its replay failed and the change must be delivered again
+ */
+ public void setOwned(boolean owned)
+ {
+ this.owned = owned;
+ }
+
+ /**
+ * Records that the replay of this change failed once more.
+ *
+ * @param nowMs
+ * when it failed, on a clock which only moves forward
+ */
+ public void recordReplayFailure(long nowMs)
+ {
+ if (replayFailures == 0)
+ {
+ firstReplayFailureTimeMs = nowMs;
+ }
+ replayFailures++;
+ }
+
+ /**
+ * Returns how many times in a row the replay of this change failed.
+ *
+ * @return the number of failures, 0 when its replay never failed
+ */
+ public int getReplayFailures()
+ {
+ return replayFailures;
+ }
+
+ /**
+ * Returns how long the replay of this change has been failing.
+ *
+ * @param nowMs
+ * the current time, on the clock {@link #recordReplayFailure(long)} was given
+ * @return the duration in milliseconds, 0 when its replay never failed
+ */
+ public long getReplayFailingForMs(long nowMs)
+ {
+ return replayFailures == 0 ? 0 : nowMs - firstReplayFailureTimeMs;
+ }
+
+ /**
* Get the operation associated to the PendingChange.
* @return the operation
*/
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 06d3797..4eef2d8 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;
@@ -81,6 +82,19 @@
private final ServerState state;
/**
+ * How many of the changes listed here have a replay failure recorded against them.
+ * <p>
+ * A change is counted from its first failed replay until it leaves this map, and it is
+ * what tells a change which can not be applied from a backend which is serving again:
+ * an outage fails everything in flight, while a change which can never be applied here
+ * fails alone, among changes which replay perfectly well. The session restart backoff
+ * reads it, so that the successful replays which surround a failing change do not keep
+ * resetting the wait this domain has reached on it (issue #889).
+ */
+ @GuardedBy("pendingChangesLock")
+ private int failingChanges;
+
+ /**
* Creates a new RemotePendingChanges using the provided ServerState.
*
* @param state The ServerState that will be updated when LDAPUpdateMsg
@@ -111,6 +125,10 @@
/**
* Returns the number of changes actively being replayed.
+ * <p>
+ * A change whose replay failed counts here until it is applied or given up on: it stays
+ * a dependency of the changes which follow it, since a change which is not in the data
+ * is exactly what they must wait for, and the delivery which comes next takes it over.
*
* @return the number of changes actively being replayed.
*/
@@ -140,19 +158,48 @@
/**
* Add a new LDAPUpdateMsg that was received from the replication server
* to the pendingList.
+ * <p>
+ * A change which is already listed and which a replay thread owns - it is being
+ * replayed, it waits for the change it depends on, or it has been replayed and waits
+ * for the changes before it - is left alone: that copy knows what this one does not,
+ * and replaying it again is exactly what the duplicate check is there to prevent
+ * (OPENDJ-1115).
+ * <p>
+ * A change which is listed but which no replay thread owns is taken over by this
+ * delivery: it is the one to replay, and the copy which may still wait in the replay
+ * queue is dropped by {@link #markInProgress(LDAPUpdateMsg)}. That is a change whose
+ * replay failed - it is deliberately left out of the ServerState, so the replication
+ * server sends it again over the session which was restarted (issue #889) - and it is
+ * also, harmlessly, a change which was listed a moment ago by a delivery no replay
+ * thread has picked up yet: the two deliveries carry the same change, and the last one
+ * listed is the one replayed.
+ * <p>
+ * The failures the change went through are kept, whichever delivery replays it: they
+ * are what has this replica eventually give up on a change it can not apply.
*
* @param update The LDAPUpdateMsg that was received from the replication
* server and that will be added to the pending list.
* @return {@code false} if the update was already registered in the pending
- * changes.
+ * changes and is owned by a replay thread.
*/
public boolean putRemoteUpdate(LDAPUpdateMsg update)
{
pendingChangesWriteLock.lock();
try
{
- CSN csn = update.getCSN();
- return pendingChanges.put(csn, new PendingChange(csn, null, update)) == null;
+ final CSN csn = update.getCSN();
+ final PendingChange listed = pendingChanges.get(csn);
+ if (listed == null)
+ {
+ pendingChanges.put(csn, new PendingChange(csn, null, update));
+ return true;
+ }
+ if (listed.isCommitted() || listed.isOwned())
+ {
+ return false;
+ }
+ listed.setMsg(update);
+ return true;
}
finally
{
@@ -177,6 +224,7 @@
throw new NoSuchElementException();
}
curChange.setCommitted(true);
+ curChange.setOwned(false);
activeAndDependentChanges.remove(curChange);
final Iterator<PendingChange> it = pendingChanges.values().iterator();
@@ -191,6 +239,11 @@
{
state.update(pendingChange.getCSN());
}
+ if (pendingChange.getReplayFailures() > 0)
+ {
+ // The change is leaving this map, so it is not one of the failing ones anymore.
+ failingChanges--;
+ }
it.remove();
}
}
@@ -200,18 +253,203 @@
}
}
- public void markInProgress(LDAPUpdateMsg msg)
+ /**
+ * Gives up the ownership a replay thread had on the change with the provided CSN,
+ * whose replay failed.
+ * <p>
+ * The change stays listed here and stays uncommitted: it is the barrier which keeps
+ * the ServerState - and every change which follows it - from moving past a change
+ * which is not in the data (issue #889), and it is what has the replication server
+ * send it again over the restarted session. Only the mark which says a replay thread
+ * owns it is dropped, so that {@link #putRemoteUpdate(LDAPUpdateMsg)} takes the next
+ * delivery instead of discarding it as a duplicate.
+ * <p>
+ * It also stays listed among the changes the newer ones are checked against: a change
+ * which is not in the data yet is exactly what the changes which follow it must depend
+ * on, whether or not a replay thread owns it right now.
+ * <p>
+ * The changes another replay thread is applying right now are left alone: they are
+ * about to commit, and forgetting them would have their {@code commit()} fail, the
+ * ServerState stay behind them and the replication server replay them a second time.
+ *
+ * @param csn the CSN of the change whose replay failed
+ */
+ public void replayFailed(CSN csn)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(csn);
+ if (change != null && !change.isCommitted())
+ {
+ change.setOwned(false);
+ }
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /** How long, and how many times, the replay of one change has been failing. */
+ static final class ReplayFailure
+ {
+ private final int attempts;
+ private final long failingForMs;
+
+ private ReplayFailure(int attempts, long failingForMs)
+ {
+ this.attempts = attempts;
+ this.failingForMs = failingForMs;
+ }
+
+ /**
+ * Returns how many deliveries of the change failed to replay it in a row. A delivery
+ * is attempted several times in place before it is counted here, so this is a count
+ * of deliveries rather than of attempts made on the backend.
+ *
+ * @return the number of failed deliveries, at least 1
+ */
+ int getAttempts()
+ {
+ return attempts;
+ }
+
+ /**
+ * Returns how long the replay of the change has been failing, that is the time
+ * between its first failure and the one which was just recorded.
+ *
+ * @return the duration in milliseconds, 0 for a first failure
+ */
+ long getFailingForMs()
+ {
+ return failingForMs;
+ }
+ }
+
+ /**
+ * Records that the replay of the change with the provided CSN failed once more.
+ * <p>
+ * The failures are kept on the change itself, which stays listed for as long as it has
+ * not been applied, so a change which keeps failing keeps its give-up budget across the
+ * deliveries which take over from one another, and a change which is replayed or given
+ * up on takes its failures away with it (issue #889).
+ *
+ * @param csn
+ * the CSN of the change whose replay failed
+ * @param nowMs
+ * when it failed, on a clock which only moves forward
+ * @return the failures of the change, or {@code null} when it is not listed as an
+ * uncommitted change anymore, which happens when the domain was disabled while
+ * it was being replayed: there is no change left here to give up on
+ */
+ public ReplayFailure recordReplayFailure(CSN csn, long nowMs)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(csn);
+ if (change == null || change.isCommitted())
+ {
+ return null;
+ }
+ if (change.getReplayFailures() == 0)
+ {
+ // Its first failure: this change joins the ones which are failing right now.
+ failingChanges++;
+ }
+ change.recordReplayFailure(nowMs);
+ return new ReplayFailure(change.getReplayFailures(), change.getReplayFailingForMs(nowMs));
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
+ * Returns whether the replay of any change listed here is failing right now.
+ * <p>
+ * A change is failing from its first failed replay until it leaves this map, whether
+ * it leaves it applied or given up on. The session restart backoff reads this: a
+ * change which was replayed only says that this backend is serving again when it is
+ * the last one which was failing, and the successful replays which surround a change
+ * this replica can not apply must not keep resetting the wait it has reached on it.
+ *
+ * @return {@code true} while at least one listed change has a failed replay recorded
+ * against it
+ */
+ boolean hasFailingChanges()
{
pendingChangesReadLock.lock();
try
{
- activeAndDependentChanges.add(pendingChanges.get(msg.getCSN()));
+ return failingChanges > 0;
}
finally
{
pendingChangesReadLock.unlock();
}
}
+
+ /**
+ * Forgets every change listed here, without updating the ServerState.
+ * <p>
+ * Called when the domain is disabled: its ServerState is saved and cleared from
+ * memory, and it is loaded again from the backend when the domain is enabled back, so
+ * the bookkeeping which goes with it must not outlive it. A change which stayed here
+ * would be discarded as a duplicate when the replication server sends it again, and
+ * nothing would ever replay it or record it in the ServerState.
+ */
+ public void clear()
+ {
+ pendingChangesWriteLock.lock();
+ dependentChangesLock.lock();
+ try
+ {
+ pendingChanges.clear();
+ dependentChanges.clear();
+ activeAndDependentChanges.clear();
+ failingChanges = 0;
+ }
+ 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 replication
+ * server delivered the change again and that delivery took over from this one,
+ * or the domain was disabled and forgot the change, so this copy must not be
+ * replayed.
+ */
+ public boolean markInProgress(LDAPUpdateMsg msg)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(msg.getCSN());
+ if (change == null || change.isCommitted() || change.getLDAPUpdateMsg() != msg)
+ {
+ return false;
+ }
+ change.setOwned(true);
+ activeAndDependentChanges.add(change);
+ return true;
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
/**
* Get the first update in the list that have some dependencies cleared.
*
@@ -223,7 +461,7 @@
dependentChangesLock.lock();
try
{
- if (!dependentChanges.isEmpty())
+ if (!dependentChanges.isEmpty() && !pendingChanges.isEmpty())
{
PendingChange firstDependentChange = dependentChanges.first();
if (pendingChanges.firstKey().isNewerThanOrEqualTo(firstDependentChange.getCSN()))
@@ -248,14 +486,25 @@
*/
private void addDependency(PendingChange dependentChange)
{
+ pendingChangesReadLock.lock();
dependentChangesLock.lock();
try
{
- dependentChanges.add(dependentChange);
+ /*
+ * A change which is not listed as pending anymore - the domain was disabled while
+ * its dependencies were computed - must not be listed as dependent either:
+ * getNextUpdate() reads both and would hand out a change nothing owns. The
+ * replication server sends it again when the domain is enabled back.
+ */
+ if (pendingChanges.containsKey(dependentChange.getCSN()))
+ {
+ dependentChanges.add(dependentChange);
+ }
}
finally
{
dependentChangesLock.unlock();
+ pendingChangesReadLock.unlock();
}
}
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 44ea968..4a31d0c 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 @@
// 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/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 73bf3c2..03f5309 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
@@ -697,7 +697,7 @@
}
/** Increment the number of processed updates. */
- private void incProcessedUpdates()
+ protected final void incProcessedUpdates()
{
numProcessedUpdates.incrementAndGet();
}
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 ff7f3f1..999bef3 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 @@
"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 d92dacc..cde9f7a 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -615,3 +615,12 @@
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" (delivery %d, each \
+ attempted several times in place). 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": its replay has been \
+ failing for %d ms over %d deliveries, each attempted several times in place. The change is being \
+ skipped: this replica now diverges from the rest of the topology and must be reinitialized
+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
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 181b691..f395262 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 @@
/** 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<String, AtomicInteger> alertCountByType = new ConcurrentHashMap<>();
+
/** Creates a new instance of this SMTP alert handler. */
public DummyAlertHandler()
{
@@ -86,6 +92,7 @@
LocalizableMessage alertMessage)
{
alertCount.incrementAndGet();
+ alertCountByType.computeIfAbsent(alertType, type -> new AtomicInteger()).incrementAndGet();
}
/**
@@ -98,6 +105,21 @@
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 2a04473..751daac 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 @@
}
// 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,28 @@
/** Registered short circuits for operations regardless of controls. */
private static Map<String, Integer> shortCircuits = new ConcurrentHashMap<>();
+ /** How many times a registered short circuit was reached. */
+ private static final Map<String, AtomicInteger> shortCircuitCounts = new ConcurrentHashMap<>();
+
+ /** How many times a registered short circuit must be applied, when it is limited. */
+ private static final Map<String, Integer> 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;
+ }
+
+
/**
* Register a short circuit for the given operation type and plugin point.
* @param operation The type of operation the short circuit applies to.
@@ -670,7 +701,29 @@
*/
public static void registerShortCircuit(OperationType operation, String section, int resultCode)
{
- shortCircuits.put(operation + "/" + section.toLowerCase(), resultCode);
+ final String key = operation + "/" + section.toLowerCase();
+ // This registration applies to every operation, and it counts from zero: a limit or
+ // a count left behind by a previous registration is not part of it.
+ shortCircuitCounts.remove(key);
+ shortCircuitLimits.remove(key);
+ shortCircuits.put(key, 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);
}
/**
@@ -680,6 +733,11 @@
*/
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);
+ // The count belongs to the registration which is being removed: a test which counts
+ // the operations it short circuits must not inherit the count of the previous one.
+ shortCircuitCounts.remove(key);
}
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
index 54d858c..0a2028f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
@@ -102,6 +102,40 @@
/** Generation id for a fully empty domain. */
public static final long EMPTY_DN_GENID = GenerationIdChecksum.EMPTY_BACKEND_GENERATION_ID;
+ /** How many times {@link #assertMonitorAttrValueStays} reads a value by default. */
+ private static final int MONITOR_ATTR_SAMPLES = 5;
+
+ /** How long {@link #assertMonitorAttrValueStays} waits between two reads. */
+ private static final long MONITOR_ATTR_SAMPLE_INTERVAL_IN_MS = 200;
+
+ /**
+ * How long {@link #assertMonitorAttrValueStays} waits for the monitor entry of a domain
+ * to be registered again before it gives up on reading it: longer than the
+ * {@code MAX_REPLAY_RETRY_DELAY_IN_MS} a session restart holds it down for.
+ */
+ private static final long MONITOR_ATTR_SAMPLE_GRACE_IN_MS = 30000;
+
+ /**
+ * How much longer than the samples it asks for {@link #assertMonitorAttrValueStays}
+ * runs before it gives up: the samples of a domain which keeps restarting its session
+ * are taken a restart apart, and waiting for all of them would outlast the fork.
+ */
+ private static final long MONITOR_ATTR_SAMPLES_DEADLINE_IN_MS = 60000;
+
+ /**
+ * How many samples it takes for {@link #assertMonitorAttrValueStays} to outlast the
+ * session restart which brings a change back, so that a counter only that delivery
+ * could bump a second time is watched while it arrives.
+ * <p>
+ * It covers a domain which restarts its session for the first time, which waits
+ * {@code LDAPReplicationDomain.REPLAY_RETRY_DELAY_IN_MS} before reconnecting. The wait
+ * of a domain which has been restarting its session in a row is longer - it climbs to
+ * {@code MAX_REPLAY_RETRY_DELAY_IN_MS} - so a redelivery is outside this window there;
+ * sampling for ten seconds at every call site to cover it would cost more than the
+ * assertions are worth.
+ */
+ protected static final int MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY = 12;
+
/** The internal connection used for operation. */
protected InternalClientConnection connection;
@@ -484,17 +518,170 @@
@Override
public Long call() throws Exception
{
- String monitorFilter = "(&(cn=Directory server*)(domain-name=" + baseDN + "))";
- InternalSearchOperation op =
- connection.processSearch(newSearchRequest("cn=replication,cn=monitor", WHOLE_SUBTREE, monitorFilter));
- Assertions.assertThat(op.getSearchEntries()).as("Could not read monitoring information").isNotEmpty();
-
- SearchResultEntry entry = op.getSearchEntries().getFirst();
- return entry.parseAttribute(attr).asLong();
+ Long value = readMonitorAttrValue(baseDN, attr);
+ Assertions.assertThat(value)
+ .as("the monitor entry of %s is not registered", baseDN).isNotNull();
+ return value;
}
});
}
+ /**
+ * Reads a monitor attribute of a replication domain, once.
+ *
+ * @param baseDN the base DN of the domain whose monitor entry to read
+ * @param attr the monitor attribute to read
+ * @return the value of the attribute, or {@code null} when the monitor entry of the
+ * domain is not registered - which it is not for as long as its session to the
+ * replication server is down
+ * @throws Exception if the monitor could not be searched, or if the entry is there and
+ * does not publish the attribute, which is a wrong name rather than
+ * something to wait for
+ */
+ private Long readMonitorAttrValue(final DN baseDN, final String attr) throws Exception
+ {
+ String monitorFilter = "(&(cn=Directory server*)(domain-name=" + baseDN + "))";
+ InternalSearchOperation op =
+ connection.processSearch(newSearchRequest("cn=replication,cn=monitor", WHOLE_SUBTREE, monitorFilter));
+ if (op.getSearchEntries().isEmpty())
+ {
+ return null;
+ }
+ SearchResultEntry entry = op.getSearchEntries().getFirst();
+ Long value = entry.parseAttribute(attr).asLong();
+ Assertions.assertThat(value)
+ .as("the monitor entry of %s does not publish %s", baseDN, attr).isNotNull();
+ return value;
+ }
+
+ /**
+ * Waits for a monitor attribute of a replication domain to reach the expected value.
+ * <p>
+ * The monitor entry of a domain is deregistered for as long as its session to the
+ * replication server is down, which is what a replay failure does to it, and a counter
+ * is bumped a moment after the change or the delivery it counts was dealt with:
+ * reading the value once would be a race on both counts.
+ * <p>
+ * The read is deliberately {@link #readMonitorAttrValue(DN, String)} rather than the
+ * retrying {@link #getMonitorAttrValue(DN, String)}: a {@link TestTimer} budget is a
+ * number of steps rather than a deadline, so one timer waiting on another multiplies
+ * them - 150 steps around a read which sleeps ten seconds of its own is 25 minutes,
+ * long past the {@code org.opends.test.timeout} the fork is killed on. One timer owns
+ * the deadline here, and a monitor entry which is not registered is one failed poll -
+ * so the deadline has to be wide enough for a domain which is restarting its session to
+ * register it again, which takes the backoff of that restart.
+ *
+ * @param baseDN the base DN of the domain whose monitor entry to read
+ * @param attributeName the monitor attribute to read
+ * @param expected the value it must reach
+ * @param message what is being asserted
+ * @throws Exception if the value was not reached in time
+ */
+ protected void assertMonitorAttrValueEventually(
+ final DN baseDN, final String attributeName, final long expected, final String message)
+ throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertEquals(readMonitorAttrValue(baseDN, attributeName), (Long) expected, message);
+ }
+ });
+ }
+
+ /**
+ * Checks that a monitor attribute of a replication domain holds the expected value and
+ * keeps holding it, over {@link #MONITOR_ATTR_SAMPLES} samples.
+ *
+ * @param baseDN the base DN of the domain whose monitor entry to read
+ * @param attributeName the monitor attribute to read
+ * @param expected the value it must hold
+ * @param message what is being asserted
+ * @throws Exception if the value changes, or if the monitor entry can not be read
+ */
+ protected void assertMonitorAttrValueStays(
+ final DN baseDN, final String attributeName, final long expected, final String message)
+ throws Exception
+ {
+ assertMonitorAttrValueStays(baseDN, attributeName, expected, MONITOR_ATTR_SAMPLES, message);
+ }
+
+ /**
+ * Checks that a monitor attribute of a replication domain holds the expected value and
+ * keeps holding it, over the provided number of samples.
+ * <p>
+ * Waiting for a value to be reached is not enough to tell that something happened only
+ * once: a counter which is bumped a second time goes through the expected value on its
+ * way, and the first poll which sees it passes.
+ * <p>
+ * The samples have to outlast whatever could bump the counter a second time, or the
+ * assertion only reads like it is watching for it. The default is enough for a second
+ * attempt of the same delivery, which is fifty milliseconds away; a counter which a
+ * change delivered again could bump has to be watched for longer than the session
+ * restart which brings that delivery, so those call sites pass
+ * {@link #MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY}.
+ * <p>
+ * The monitor entry of a domain is gone for as long as its session is down, which a
+ * session restart in the middle of the window does: a read which comes back with
+ * nothing is not a sample rather than a failure, and the samples asked for are taken
+ * once it is back. So a restart stretches the window rather than shortening it, which
+ * is the right way round for what is being asserted, and the entry staying away for
+ * {@link #MONITOR_ATTR_SAMPLE_GRACE_IN_MS} is what fails the assertion. The samples are
+ * bounded all the same: a domain which restarts its session over and over would
+ * otherwise have this wait for one readable moment per restart until the fork is killed
+ * for taking too long, which says nothing about the value being watched.
+ *
+ * @param baseDN the base DN of the domain whose monitor entry to read
+ * @param attributeName the monitor attribute to read
+ * @param expected the value it must hold
+ * @param samples how many times to read the value, at least
+ * {@link #MONITOR_ATTR_SAMPLE_INTERVAL_IN_MS} apart
+ * @param message what is being asserted
+ * @throws Exception if the value changes, or if the monitor entry can not be read
+ */
+ protected void assertMonitorAttrValueStays(final DN baseDN, final String attributeName,
+ final long expected, final int samples, final String message) throws Exception
+ {
+ final long now = System.currentTimeMillis();
+ final long deadline = now + samples * MONITOR_ATTR_SAMPLE_INTERVAL_IN_MS
+ + MONITOR_ATTR_SAMPLES_DEADLINE_IN_MS;
+ long readableBy = now + MONITOR_ATTR_SAMPLE_GRACE_IN_MS;
+ int taken = 0;
+ while (taken < samples)
+ {
+ final Long value = readMonitorAttrValue(baseDN, attributeName);
+ if (value != null)
+ {
+ assertEquals(value, (Long) expected, message);
+ taken++;
+ if (taken == samples)
+ {
+ // Every sample which was asked for held the value: how long they took to take
+ // is not what this is asserting.
+ return;
+ }
+ readableBy = System.currentTimeMillis() + MONITOR_ATTR_SAMPLE_GRACE_IN_MS;
+ }
+ else if (System.currentTimeMillis() > readableBy)
+ {
+ fail("the monitor entry of " + baseDN + " was not registered again in "
+ + MONITOR_ATTR_SAMPLE_GRACE_IN_MS + "ms: " + message);
+ }
+ if (System.currentTimeMillis() > deadline)
+ {
+ fail("only " + taken + " of " + samples + " samples of " + attributeName
+ + " could be read before the deadline: " + message);
+ }
+ Thread.sleep(MONITOR_ATTR_SAMPLE_INTERVAL_IN_MS);
+ }
+ }
+
protected void checkEntryHasAttributeValue(final DN dn, final String attrTypeStr, final String valueString,
int timeoutInSecs, String notFoundErrorMsg) throws Exception
{
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 4c9dde1..b9ce72b 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;
@@ -35,7 +36,9 @@
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
+import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.DecodeException;
import org.forgerock.opendj.ldap.ModificationType;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.requests.ModifyDNRequest;
@@ -46,25 +49,33 @@
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ModifyOperation;
+import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.extensions.DummyAlertHandler;
import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.protocols.internal.InternalClientConnection;
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;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyContext;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
+import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.Attribute;
import org.opends.server.types.Attributes;
import org.opends.server.types.Entry;
+import org.opends.server.types.LDAPException;
import org.opends.server.types.Modification;
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
+import org.opends.server.types.RawModification;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
import org.opends.server.util.TimeThread;
@@ -81,6 +92,13 @@
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /**
+ * How long a change is retried in the tests which check that this replica gives up on
+ * a change it can never apply: long enough for the change to be delivered again a
+ * couple of times, short enough not to make the test wait out a real backend outage.
+ */
+ private static final long TEST_GIVE_UP_DELAY_IN_MS = 2000;
+
/** An entry with a entryUUID. */
private Entry personWithUUIDEntry;
private Entry personWithSecondUniqueID;
@@ -1376,6 +1394,716 @@
}
/**
+ * 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);
+ final long giveUpDelay = domain.getReplayGiveUpDelay();
+ try
+ {
+ // A backend which is down for maintenance is waited out for minutes: this test
+ // can not, so the change is given up on after a couple of deliveries instead.
+ // Set inside the try which puts it back, like the short circuit below: both are
+ // the domain's and the server's for as long as they are left behind.
+ domain.setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_IN_MS);
+ /*
+ * 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.DELETE, "PreParse", ResultCode.OTHER.intValue());
+
+ 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 delivered more than once is what tells that it was not recorded as
+ * replayed.
+ *
+ * One delivery is retried in place IN_PLACE_REPLAY_ATTEMPTS times before the
+ * session is restarted, so it takes more than that many short circuits to prove
+ * that the change was delivered a second time.
+ */
+ 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")
+ > IN_PLACE_REPLAY_ATTEMPTS,
+ "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");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be replayed must be counted once, not once per attempt");
+ /*
+ * A counter bumped once per attempt rather than once per change goes through the
+ * expected value on its way, so the value has to be seen to stay put rather than
+ * to be reached once.
+ */
+ assertMonitorAttrValueStays(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");
+ domain.setReplayGiveUpDelay(giveUpDelay);
+ }
+ }
+ 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.
+ * <p>
+ * 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");
+ final long giveUpDelay = domain.getReplayGiveUpDelay();
+ try
+ {
+ // Both are put back by the finally below, so both are set inside the try.
+ domain.setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_IN_MS);
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+
+ 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");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 2,
+ "both changes must be counted as failed, once each");
+ /*
+ * Two changes counted more than once each climb past +2, and the poll which lands
+ * on it would pass: the value has to be seen to stay put.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 2,
+ "both changes must be counted as failed, once each");
+ 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");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ domain.setReplayGiveUpDelay(giveUpDelay);
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * The result codes a replay is retried on rather than skipped: the storage failing to
+ * serve the operation, and a lock which could not be taken (OPENDJ-885) - the ten
+ * in-place attempts only yield to the thread holding it, so a lock held for a while
+ * burns every one of them and the change is as absent from the data as after a storage
+ * failure.
+ */
+ @DataProvider(name = "transientReplayFailures")
+ public Object[][] transientReplayFailures()
+ {
+ return new Object[][] {
+ { ResultCode.UNAVAILABLE, 14, "user.889.3" },
+ { ResultCode.BUSY, 15, "user.889.4" },
+ };
+ }
+
+ /**
+ * Test case for [Issue 889]: a replay which fails on the server itself has the session
+ * restarted and the change delivered again, and a failure which clears in the meantime
+ * has the change applied exactly once, without the change being given up on and without
+ * it being reported as failed.
+ */
+ @Test(dataProvider = "transientReplayFailures")
+ public void transientReplayFailureIsRetriedAndTheChangeApplied(
+ final ResultCode transientFailure, final int serverId, final String uid) throws Exception
+ {
+ testSetUp("transientReplayFailureIsRetriedAndTheChangeApplied." + uid);
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : transientReplayFailureIsRetriedAndTheChangeApplied "
+ + transientFailure));
+
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=" + uid + "," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: " + uid,
+ "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 long initialReplayed = getMonitorAttrValue(baseDN, "replayed-updates-ok");
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ /*
+ * The backend is unavailable the way it is while a rebuild is performed or while it
+ * is offline (OPENDJ-49), and it stays unavailable for longer than the replay is
+ * retried in place: the change is only applied if the session is restarted and the
+ * replication server delivers it a second time.
+ */
+ try
+ {
+ // Registered inside the try which deregisters it: the plugin is consulted for
+ // every delete in this server, so one left behind fails the tests which follow.
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ transientFailure.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+
+ 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 change must have been delivered again after the session was restarted")
+ .isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
+
+ 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");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1,
+ "the change must be recorded as replayed");
+ /*
+ * A change applied twice - the delivery which failed and the one which took over
+ * from it, the OPENDJ-1115 regression the takeover is there to prevent - takes the
+ * counter through +1 on its way to +2, so the value has to be seen to stay put
+ * rather than to be reached once. It has to be watched for longer than the
+ * session restart which brings that second delivery, too, or the assertion stops
+ * looking before the delivery it is looking for could arrive.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-ok", initialReplayed + 1,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was delivered again must be applied exactly once");
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "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();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: the result code the server puts on an internal error is
+ * configurable and is not validated as a result code, so it can be set to one conflict
+ * resolution knows how to solve. Such a change is left to conflict resolution, and when
+ * that can not solve it either the change is retried as the storage failure it is -
+ * recording it as replayed after one attempt would be issue #889 again.
+ */
+ @Test
+ public void changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried() throws Exception
+ {
+ testSetUp("changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried"));
+
+ final int serverId = 16;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.5," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.5",
+ "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);
+
+ /*
+ * UNWILLING_TO_PERFORM is one of the codes solveNamingConflict(ModifyDNOperation)
+ * solves, so it must not be treated as a failure of the server before conflict
+ * resolution had its chance - and it is what the storage reports here.
+ */
+ // Put back whatever was configured, not the default: a suite which runs with
+ // another server-error-result-code must not be rewritten by this test.
+ final int previousServerErrorResultCode =
+ getServerContext().getCoreConfigManager().getServerErrorResultCode().intValue();
+ try
+ {
+ /*
+ * Changed inside the try which puts it back: the result code this server reports
+ * an internal error with is server-wide, so one left behind would change which
+ * road every later replay of this suite takes.
+ */
+ setServerErrorResultCode(ResultCode.UNWILLING_TO_PERFORM.intValue());
+ /*
+ * The failure lasts longer than the attempts made in place, so the change is only
+ * applied if it was left out of the ServerState and delivered again rather than
+ * recorded as replayed once conflict resolution reported it could not be solved.
+ */
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ ResultCode.UNWILLING_TO_PERFORM.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ assertNull(getEntry(tmp.getName(), 120000, false),
+ "the change was skipped rather than retried once the storage served the operation");
+ Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
+ .as("the change must have been delivered again rather than recorded as replayed")
+ .isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was replayed in the end must not be counted as given up on");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a change which was replayed in the end must not tell the administrator that this replica diverged");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ setServerErrorResultCode(previousServerErrorResultCode);
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: a change whose message can not be turned into an operation
+ * must not hold this replica's ServerState back for good.
+ * <p>
+ * There is no operation to retry and no delivery which would decode any better, so the
+ * change has to be skipped rather than left listed as the barrier: a change which stays
+ * uncommitted holds back the ServerState - and every change which follows it, from
+ * every master - and the delivery which would replace it is turned down while a replay
+ * thread still owns it, so nothing would ever move it again.
+ */
+ @Test
+ public void aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack() throws Exception
+ {
+ testSetUp("aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack"));
+
+ final int serverId = 17;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.6," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.6",
+ "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);
+
+ final CSN csn = gen.newCSN();
+ broker.publish(undecodableModifyMsg(csn, tmp.getName(), uuid));
+
+ 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),
+ "a change which can never be decoded must not hold the ServerState back");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be decoded must be counted as failed");
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be decoded must be counted once");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Builds a ModifyMsg which travels the protocol intact and can not be turned into an
+ * operation.
+ * <p>
+ * The encoded modifications are carried as an opaque byte array and are only read by
+ * {@code createOperation()}, so a message whose modifications are corrupt is decoded,
+ * listed as pending and handed to a replay thread before it fails - which is the point
+ * of this test.
+ *
+ * @param csn the CSN to give the change
+ * @param dn the entry the change is on
+ * @param entryUUID the UUID of that entry
+ * @return a message whose replay can not build an operation
+ * @throws Exception if the message could not be built
+ */
+ private ModifyMsg undecodableModifyMsg(CSN csn, DN dn, String entryUUID) throws Exception
+ {
+ final List<Modification> mods = generatemods("description", "the decoding must fail here");
+ final byte[] bytes =
+ new ModifyMsg(csn, dn, mods, entryUUID).getBytes(ProtocolVersion.getCurrentVersion());
+
+ /*
+ * Break the length of the attribute description inside the encoded modifications, so
+ * that the ASN.1 reader runs past the end of them. The attribute name only appears
+ * there, and the byte before it is the length it is read with.
+ */
+ final int attributeName = indexOf(bytes, "description".getBytes("UTF-8"));
+ assertTrue(attributeName > 0, "the encoded modifications must carry the attribute name");
+ bytes[attributeName - 1] = (byte) 0x7F;
+
+ final ModifyMsg corrupted =
+ (ModifyMsg) ReplicationMsg.generateMsg(bytes, ProtocolVersion.getCurrentVersion());
+ try
+ {
+ corrupted.createOperation(getRootConnection());
+ fail("this test needs a message which can not be turned into an operation");
+ }
+ catch (LDAPException | DecodeException expected)
+ {
+ /*
+ * Which is what the replay of this message hits: the ASN.1 reader reports a
+ * DecodeException, which RawModification.decode() reports as an LDAPException and
+ * ModifyCommonMsg.decodeRawMods() lets through as it is when the over-read lands
+ * between two modifications rather than inside one. The two are named rather than
+ * caught as an Exception so that this test says what the message does, but neither
+ * is what decides its fate: this change is given up on because no operation could
+ * be built from it, and a failure of an operation which was built takes the other
+ * road whatever it was thrown as, which
+ * aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed pins.
+ */
+ }
+ return corrupted;
+ }
+
+ /**
+ * Test case for [Issue 889]: a change whose operation was built is delivered again
+ * rather than recorded as replayed when the replay fails before that operation could
+ * tell which change it carries.
+ * <p>
+ * Which of the two roads a failure takes is decided by the operation rather than by
+ * its CSN: a message no operation could be built from will not build one on the next
+ * delivery either, so it is given up on where it is reported, while an operation which
+ * was built may well have reached the backend - so its change is kept out of the
+ * ServerState and asked for again, wherever in the replay the failure happened. The
+ * entry DN of a ModifyMsg which does not parse is that case: it leaves
+ * {@code getEntryDN()} null and the replay throws before the CSN of the operation is
+ * read, so a give-up keyed off that CSN would record a change which never reached the
+ * backend as replayed, which is this issue by another route.
+ */
+ @Test
+ public void aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed() throws Exception
+ {
+ testSetUp("aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed"));
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.7," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.7",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final DN dn = tmp.getName();
+ final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ final CSNGenerator gen = new CSNGenerator(18, TimeThread.getTime());
+ final CSN csn = gen.newCSN();
+ final String description = "the replay must fail once the operation is built";
+ final List<Modification> mods = generatemods("description", description);
+
+ domain.processUpdate(new ModifyMsgWithAnUnparseableOperationDN(csn, dn, mods, uuid));
+
+ /*
+ * Long enough to outlast the session restart the failure asks for: a change which is
+ * being asked for again is not in the data at any point of it.
+ */
+ for (int i = 0; i < MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY; i++)
+ {
+ assertFalse(domain.getServerState().cover(csn),
+ "a change whose operation was built must be asked for again, not recorded as replayed");
+ Thread.sleep(200);
+ }
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ "a change which is still to be delivered again must not be counted as given up on");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a change which is still to be delivered again must not be alerted on as a divergence");
+
+ /*
+ * The failed change is the barrier which holds this domain's ServerState back until
+ * it is replayed, and the replication server sending it again is what replays it.
+ * Nothing sends this one - it never travelled a session - so the delivery which takes
+ * over from the one which failed is made here, and it is made until it is taken: a
+ * delivery is dropped rather than queued while the listener thread is down, which it
+ * is for as long as the recovery is restarting the session, and the monitor entry
+ * read above comes back with the broker rather than with the listener. A delivery of
+ * a change a replay thread owns is refused as the duplicate it is, and the ServerState
+ * keeps this from delivering a change which was replayed a second time.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(csn))
+ {
+ domain.processUpdate(new ModifyMsg(csn, dn, mods, uuid));
+ }
+ assertTrue(domain.getServerState().cover(csn),
+ "the change must be recorded as replayed once it has been delivered again");
+ }
+ });
+ checkEntryHasAttributeValue(dn, "description", description, 30,
+ "the change must be applied by the delivery which took over from the failed one");
+ }
+
+ /**
+ * 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
+ * on - and its entry DN does not parse, which is what has
+ * {@code ModifyOperationBasis.getEntryDN()} return null and the replay throw before
+ * {@code OperationContext.getCSN(op)} is reached. Such a message can not travel the
+ * protocol: the DN of a ModifyMsg is decoded on the way in and the operation is built
+ * from its {@code toString()}, so this one is handed to the domain rather than
+ * published.
+ */
+ private static final class ModifyMsgWithAnUnparseableOperationDN extends ModifyMsg
+ {
+ private ModifyMsgWithAnUnparseableOperationDN(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
+ {
+ final ModifyOperation op = new ModifyOperationBasis(connection, nextOperationID(),
+ nextMessageID(), null, ByteString.valueOfUtf8("this is not a DN"),
+ new ArrayList<RawModification>());
+ op.setAttachment(OperationContext.SYNCHROCONTEXT,
+ new ModifyContext(getCSN(), getEntryUUID()));
+ return op;
+ }
+ }
+
+ /**
+ * Returns the offset of the first occurrence of {@code needle} in {@code haystack}, or
+ * -1 when it does not occur.
+ */
+ private static int indexOf(byte[] haystack, byte[] needle)
+ {
+ for (int i = 0; i <= haystack.length - needle.length; i++)
+ {
+ int j = 0;
+ while (j < needle.length && haystack[i + j] == needle[j])
+ {
+ j++;
+ }
+ if (j == needle.length)
+ {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Sets the result code this server puts on an internal error, the way an administrator
+ * would.
+ *
+ * @param resultCode the numeric result code
+ * @throws Exception if the configuration could not be changed
+ */
+ private void setServerErrorResultCode(int resultCode) throws Exception
+ {
+ assertEquals(TestCaseUtils.applyModifications(true,
+ "dn: cn=config",
+ "changetype: modify",
+ "replace: ds-cfg-server-error-result-code",
+ "ds-cfg-server-error-result-code: " + resultCode), 0,
+ "the server error result code could not be changed");
+ }
+
+ /**
* Enable or disable the receive status of a synchronization provider.
*
* @param syncConfigDN The DN of the synchronization provider configuration
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 f59bce4..a657149 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,95 @@
return;
}
+ } finally
+ {
+ endTest(testcase);
+ }
+ }
- /* Send an update with error from the RS and get the ack with error */
+ /**
+ * 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);
- // Make the RS send a not possible assured add message
+ safeReadDomainCfgEntry = createAssuredDomain(AssuredMode.SAFE_READ_MODE, 0, TIMEOUT);
+ waitForConnectionToRs(testcase, replicationServer);
- // 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...
+ 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));
- // Check that DS replied an ack with errors
-// assertFalse(ackMsg.hasTimeout());
-// assertTrue(ackMsg.hasReplayError());
-// assertFalse(ackMsg.hasWrongStatus());
-// List<Integer> failedServers = ackMsg.getFailedServers();
-// assertEquals(failedServers.size(), 1);
-// assertEquals((integer)failedServers.get(0), (integer)1);
+ final LDAPReplicationDomain domain =
+ MultimasterReplication.findDomain(DN.valueOf(SAFE_READ_DN), null);
+ final long giveUpDelay = domain.getReplayGiveUpDelay();
+ try
+ {
+ /*
+ * 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. It is registered here rather than before the try,
+ * so that the finally below takes it back whatever happens next: the plugin is
+ * consulted for every add in this server, so one left registered would fail the
+ * fixtures of the tests which follow.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.ADD, "PreParse", ResultCode.OTHER.intValue());
+ /*
+ * A change which keeps failing is asked for again over a restarted session until
+ * this replica gives up on it. That is right, and it is not what this test is
+ * about: the FakeReplicationServer is not built to be reconnected to, and the
+ * restarts would run on while the assertions and the teardown below take their
+ * course. A give-up delay of zero has the first failure spend the whole budget, so
+ * the change is given up on where it is reported and no session is restarted: the
+ * ack of this delivery is published either way - it is sent before the give-up is
+ * decided - and the domain settles instead of reconnecting.
+ */
+ domain.setReplayGiveUpDelay(0);
+ AckMsg ackMsg = replicationServer.sendAssuredAddMsg(entry, parentUid);
+
+ assertNull(DirectoryServer.getEntry(entry.getName()), "the entry must not have been added");
+
+ // 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);
+
+ /*
+ * The change is counted as failed once the give-up is decided, which is after the
+ * ack above was published: waiting for that count is what makes the assured
+ * counters below safe to read, since they are bumped just after the publish.
+ */
+ final DN baseDN = DN.valueOf(SAFE_READ_DN);
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", 1,
+ "the change which could not be replayed must be counted as failed");
+ new MonitorAssertions(baseDN)
+ .assertValue("assured-sr-received-updates", 1)
+ .assertValue("assured-sr-received-updates-not-acked", 1)
+ .assertRemainingValuesAreZero();
+ // The failure is this replica's own, and it is in the ack it sent: nothing this
+ // domain sent went unacknowledged.
+ assertNoServerErrors(baseDN);
+ }
+ finally
+ {
+ domain.setReplayGiveUpDelay(giveUpDelay);
+ 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 0000000..5f03553
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -0,0 +1,476 @@
+/*
+ * 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.DirectoryServerTestCase;
+import org.opends.server.TestCaseUtils;
+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.BeforeClass;
+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")
+@Test(groups = { "precommit", "replication" }, sequential = true)
+public class RemotePendingChangesTest extends DirectoryServerTestCase
+{
+ private static final int SERVER_ID = 42;
+
+ @BeforeClass
+ public void startServer() throws Exception
+ {
+ // The messages these tests are built from carry DNs, which need the schema.
+ TestCaseUtils.startServer();
+ }
+
+ @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);
+ }
+
+ /**
+ * A change whose replay failed stays listed and uncommitted - it is the barrier which
+ * holds the ServerState back - but no replay thread owns it anymore, so the delivery
+ * the replication server makes over the restarted session takes over from the one
+ * which failed.
+ */
+ @Test
+ public void replayFailedLetsTheNextDeliveryTakeOverTheChange() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg failedDelivery = deleteMsg(csn, "uuid-1");
+ final DeleteMsg nextDelivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(failedDelivery));
+ assertTrue(pendingChanges.markInProgress(failedDelivery));
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "a change a replay thread owns must not be taken over (OPENDJ-1115)");
+
+ pendingChanges.replayFailed(csn);
+
+ assertEquals(pendingChanges.getQueueSize(), 1, "the change must stay listed as pending");
+ assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed");
+ assertEquals(pendingChanges.changesInProgressSize(), 1,
+ "a change which is not in the data yet must stay a dependency of the changes which follow it");
+
+ assertTrue(pendingChanges.putRemoteUpdate(nextDelivery), "the next delivery must be replayed");
+ assertFalse(pendingChanges.markInProgress(failedDelivery),
+ "the delivery whose replay failed must not be replayed again");
+ assertTrue(pendingChanges.markInProgress(nextDelivery));
+
+ pendingChanges.commit(csn);
+
+ assertTrue(state.cover(csn));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * The change whose replay failed must keep holding back the changes which follow it,
+ * including the ones which are replayed while it is being asked for again: the
+ * ServerState is a watermark, so recording any of them would record the failed change
+ * with them (issue #889).
+ */
+ @Test
+ public void replayFailedKeepsTheChangeAsABarrier() 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 inProgress = generator.newCSN();
+ final DeleteMsg failedMsg = deleteMsg(failed, "uuid-1");
+ final DeleteMsg inProgressMsg = deleteMsg(inProgress, "uuid-2");
+
+ assertTrue(pendingChanges.putRemoteUpdate(failedMsg));
+ assertTrue(pendingChanges.putRemoteUpdate(inProgressMsg));
+ assertTrue(pendingChanges.markInProgress(failedMsg));
+ assertTrue(pendingChanges.markInProgress(inProgressMsg));
+
+ // The replay of the first change failed while the second one is still being applied.
+ pendingChanges.replayFailed(failed);
+
+ // The change which was being applied when the session was restarted commits as usual.
+ pendingChanges.commit(inProgress);
+
+ assertFalse(state.cover(failed), "the change which failed must not be recorded as replayed");
+ assertFalse(state.cover(inProgress), "the failed change must hold back the ones which follow it");
+ assertEquals(pendingChanges.getQueueSize(), 2);
+
+ // The change which failed is sent again and made it this time.
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")));
+ pendingChanges.commit(failed);
+
+ assertTrue(state.cover(failed));
+ assertTrue(state.cover(inProgress));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * 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.replayFailed(csn);
+ assertTrue(pendingChanges.putRemoteUpdate(newDelivery), "the next delivery must be replayed");
+
+ assertFalse(pendingChanges.markInProgress(previousDelivery),
+ "the message of the previous delivery must not be replayed");
+ assertTrue(pendingChanges.markInProgress(newDelivery));
+ }
+
+ /**
+ * A message which was waiting in the replay queue while the domain was disabled 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.clear();
+
+ assertEquals(pendingChanges.changesInProgressSize(), 0);
+ assertFalse(pendingChanges.markInProgress(msg),
+ "a message whose change was forgotten must not be replayed");
+ }
+
+ /**
+ * A disabled domain saves its ServerState and loads it again when it is enabled back,
+ * so the changes listed as pending must not outlive it: one which stayed would be
+ * discarded as a duplicate and nothing would ever replay it.
+ */
+ @Test
+ public void clearForgetsEveryChange() 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 replayed = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(replayed, "uuid-2")));
+ pendingChanges.commit(replayed);
+
+ pendingChanges.clear();
+
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ assertEquals(pendingChanges.getDependentChangesSize(), 0);
+ assertTrue(state.isEmpty(), "forgetting the pending changes must not record them as replayed");
+ assertNull(pendingChanges.getNextUpdate());
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")),
+ "the changes must be accepted again once the domain is enabled back");
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(replayed, "uuid-2")));
+ }
+
+ /**
+ * A backend which is failing fails every change in flight: the failures of one change
+ * must not reset the ones of another, or the replica would never give up on any of
+ * them.
+ */
+ @Test
+ public void replayFailuresAreCountedPerChange() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN first = generator.newCSN();
+ final CSN second = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(second, "uuid-2")));
+
+ assertEquals(pendingChanges.recordReplayFailure(first, 1000).getAttempts(), 1);
+ assertEquals(pendingChanges.recordReplayFailure(second, 1100).getAttempts(), 1);
+ assertEquals(pendingChanges.recordReplayFailure(first, 1200).getAttempts(), 2);
+ assertEquals(pendingChanges.recordReplayFailure(second, 1300).getAttempts(), 2);
+
+ assertEquals(pendingChanges.recordReplayFailure(first, 1400).getFailingForMs(), 400);
+ assertEquals(pendingChanges.recordReplayFailure(second, 1500).getFailingForMs(), 400);
+ }
+
+ /**
+ * The failures belong to the change, which stays listed until it is applied, so they
+ * are kept across the deliveries which take over from one another: they are the budget
+ * this replica gives a change before it gives up on it, and a delivery which resets it
+ * is a replica which never gives up (issue #889).
+ */
+ @Test
+ public void replayFailuresSurviveTheDeliveryTakingOver() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg failedDelivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(failedDelivery));
+ assertTrue(pendingChanges.markInProgress(failedDelivery));
+ assertEquals(pendingChanges.recordReplayFailure(csn, 1000).getAttempts(), 1);
+ pendingChanges.replayFailed(csn);
+
+ // The replication server delivers the change again over the restarted session.
+ final DeleteMsg nextDelivery = deleteMsg(csn, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(nextDelivery));
+ assertTrue(pendingChanges.markInProgress(nextDelivery));
+
+ final RemotePendingChanges.ReplayFailure failure = pendingChanges.recordReplayFailure(csn, 301000);
+ assertEquals(failure.getAttempts(), 2);
+ assertEquals(failure.getFailingForMs(), 300000,
+ "the budget of a change must be measured from its first failure, whichever delivery failed");
+ }
+
+ /**
+ * However long a single delivery takes to fail - the replay is attempted in place
+ * several times and each attempt waits on the storage - the failures belong to the same
+ * run: a change which stops failing is applied or given up on, and it takes its
+ * failures with it.
+ */
+ @Test
+ public void replayFailuresFarApartStillBelongToTheSameRun() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ pendingChanges.recordReplayFailure(csn, 1000);
+
+ final RemotePendingChanges.ReplayFailure failure = pendingChanges.recordReplayFailure(csn, 601000);
+ assertEquals(failure.getAttempts(), 2);
+ assertEquals(failure.getFailingForMs(), 600000,
+ "a change must be given up on however long its deliveries take to fail");
+ }
+
+ /**
+ * A change which was replayed, or which the domain forgot on its way down, has no
+ * failures left to record: there is nothing left here to give up on.
+ */
+ @Test
+ public void replayFailuresGoAwayWithTheChange() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN committed = generator.newCSN();
+ final CSN forgotten = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(committed, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(forgotten, "uuid-2")));
+ assertEquals(pendingChanges.recordReplayFailure(committed, 1000).getAttempts(), 1);
+ assertEquals(pendingChanges.recordReplayFailure(forgotten, 1000).getAttempts(), 1);
+
+ pendingChanges.commit(committed);
+ assertNull(pendingChanges.recordReplayFailure(committed, 1100),
+ "a change which was replayed must not be given up on");
+
+ pendingChanges.clear();
+ assertNull(pendingChanges.recordReplayFailure(forgotten, 1100),
+ "a change a disabled domain forgot must not be given up on");
+
+ // The replication server sends it again once the domain is enabled back.
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(forgotten, "uuid-2")));
+ final RemotePendingChanges.ReplayFailure failure =
+ pendingChanges.recordReplayFailure(forgotten, 301100);
+ assertEquals(failure.getAttempts(), 1);
+ assertEquals(failure.getFailingForMs(), 0,
+ "a change which was forgotten must not be given up on straight away");
+ }
+
+ /**
+ * An outage fails every change in flight, and there are more of those than any bound a
+ * side map of failures could carry: a change must keep the budget it has been failing
+ * for however many other changes are failing with it.
+ */
+ @Test
+ public void aFailingChangeKeepsItsBudgetHoweverManyOtherChangesAreFailing() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN oldest = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(oldest, "uuid-0")));
+ assertEquals(pendingChanges.recordReplayFailure(oldest, 1000).getAttempts(), 1);
+
+ // Every other change in flight fails in between, in the order they were delivered.
+ final int changesInFlight = 1500;
+ final CSN[] others = new CSN[changesInFlight];
+ for (int i = 0; i < changesInFlight; i++)
+ {
+ others[i] = generator.newCSN();
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(others[i], "uuid-" + (i + 1))));
+ pendingChanges.recordReplayFailure(others[i], 1000 + i);
+ }
+
+ final RemotePendingChanges.ReplayFailure failure = pendingChanges.recordReplayFailure(oldest, 301000);
+ assertEquals(failure.getAttempts(), 2);
+ assertEquals(failure.getFailingForMs(), 300000,
+ "the change which has been failing the longest must be the one given up on");
+ assertEquals(pendingChanges.recordReplayFailure(others[0], 301000).getAttempts(), 2,
+ "the failures of a change must not be dropped to make room for another change");
+ }
+
+ /**
+ * The changes which are replayed around a change which keeps failing must not report
+ * that nothing is failing anymore: a change which can never be applied here fails
+ * alone, among changes which replay perfectly well, and the session restart backoff
+ * reads this to tell that apart from a backend which is serving again (issue #889).
+ */
+ @Test
+ public void aChangeKeepsFailingWhileTheChangesAroundItAreReplayed() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN failing = generator.newCSN();
+ final CSN replayed = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failing, "uuid-1")));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(replayed, "uuid-2")));
+ assertFalse(pendingChanges.hasFailingChanges(),
+ "no change has failed yet");
+
+ pendingChanges.recordReplayFailure(failing, 1000);
+ assertTrue(pendingChanges.hasFailingChanges(),
+ "the change whose replay failed must be reported as failing");
+
+ // The change which follows it is applied while the older one is still failing. It
+ // stays listed, since the ServerState can not move past the change which failed.
+ pendingChanges.commit(replayed);
+ assertTrue(pendingChanges.hasFailingChanges(),
+ "a change which was replayed must not report that the one which is failing stopped");
+
+ // Only the failing change being applied says that this backend is serving again.
+ // Both changes leave the list here: the failing one, which is the head, and the one
+ // which was waiting behind it for the ServerState to be allowed past.
+ pendingChanges.commit(failing);
+ assertFalse(pendingChanges.hasFailingChanges(),
+ "the change which was failing was applied, so nothing is failing anymore");
+
+ /*
+ * The count must be back to none, not below it: a change which is drained without
+ * ever having failed must not be counted out. A counter which went negative reads as
+ * "nothing is failing" for as long as it takes the next failures to bring it back to
+ * zero, which is what has the session restarts start over from their shortest wait
+ * while a change is failing - the very loop this counter is here to stop.
+ */
+ final CSN failingAgain = generator.newCSN();
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failingAgain, "uuid-3")));
+ pendingChanges.recordReplayFailure(failingAgain, 2000);
+ assertTrue(pendingChanges.hasFailingChanges(),
+ "the change which is failing now must be reported, whatever was drained before it");
+ }
+
+ /**
+ * The failures a repeatedly failing change accumulates are counted once, and they go
+ * away with the change however it leaves - applied, or forgotten by a disabled domain.
+ */
+ @Test
+ public void failingChangesAreCountedOnceAndForgottenWithTheChange() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN csn = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ pendingChanges.recordReplayFailure(csn, 1000);
+ pendingChanges.recordReplayFailure(csn, 2000);
+ assertTrue(pendingChanges.hasFailingChanges());
+
+ // Failing twice must not have this change counted twice, or it would still be
+ // reported as failing once it is gone.
+ pendingChanges.commit(csn);
+ assertFalse(pendingChanges.hasFailingChanges(),
+ "a change which failed several times must stop being reported as failing once");
+
+ final CSN forgotten = generator.newCSN();
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(forgotten, "uuid-2")));
+ pendingChanges.recordReplayFailure(forgotten, 3000);
+ assertTrue(pendingChanges.hasFailingChanges());
+
+ pendingChanges.clear();
+ assertFalse(pendingChanges.hasFailingChanges(),
+ "a disabled domain forgot the change, so nothing is failing here anymore");
+ }
+
+ private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception
+ {
+ return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID);
+ }
+}
--
Gitblit v1.10.0