From 9300ffec887a766b1ccf27198f69f7e5d5e3fb84 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 15 Sep 2026 07:24:43 +0000
Subject: [PATCH] [#956] Tell a failed entryUUID search apart from an entry which is not there (#968)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 385 ++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java | 555 +++++++++++++++++++++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 4
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 396 ++++++++++++++--
opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java | 85 +++
5 files changed, 1,360 insertions(+), 65 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index 537e70c..fb908f8 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
@@ -496,6 +496,23 @@
private final InternalClientConnection conn = getRootConnection();
private final AtomicBoolean shutdown = new AtomicBoolean();
private volatile boolean disabled;
+ /**
+ * Whether the data of this domain is being replaced by a total update into this replica:
+ * the import streams over the session of this domain, on its listener thread, and the
+ * backend the replay would apply a change to is deregistered for the length of it.
+ * <p>
+ * It is what {@link #disabled} is for an import or a restore run on this server, on the
+ * one road which does not set that flag: {@link #preBackendImport(LocalBackend)} takes
+ * the backend away without disabling the domain, because the domain can not stop the
+ * session it is importing over. The replay threads read it where they read
+ * {@link #disabled}: an attempt made meanwhile is made into no backend, and whatever it
+ * decided is overwritten by the import, so the change is given back instead - without a
+ * session restart, which {@link #sessionHasAnOwner()} refuses for the whole of the total
+ * update. It is set for the length of {@link #importBackend(InputStream)}, so it covers
+ * the reload of the ServerState and the reset of the pending changes which follow the
+ * import.
+ */
+ private volatile boolean importingData;
/**
* This list is used to temporary store operations that needs to be replayed
@@ -1849,7 +1866,16 @@
* this operation has already been replayed in the past.
*/
String uuid = ctx.getEntryUUID();
- if (findEntryDN(uuid) != null)
+ final DN replayedEntryDN;
+ try
+ {
+ replayedEntryDN = findEntryDN(uuid);
+ }
+ catch (SearchFailedException e)
+ {
+ return searchDidNotRun(ctx.getCSN(), e);
+ }
+ if (replayedEntryDN != null)
{
return new SynchronizationProviderResult.StopProcessing(
ResultCode.NO_OPERATION, null);
@@ -1866,7 +1892,15 @@
{
// There is a potential of perfs improvement here
// if we could avoid the following parent entry retrieval
- DN parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID());
+ final DN parentDnFromCtx;
+ try
+ {
+ parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID());
+ }
+ catch (SearchFailedException e)
+ {
+ return searchDidNotRun(ctx.getCSN(), e);
+ }
if (parentDnFromCtx == null)
{
// The parent does not exist with the specified unique id
@@ -1894,6 +1928,28 @@
}
/**
+ * Turns down an operation being replayed whose conflict resolution could not read the
+ * data, so that the change is attempted again rather than applied on what a search
+ * which did not run seemed to say.
+ * <p>
+ * The operation is reported as unavailable, which is what it is, and the search which
+ * did not run is its error message: the change is retried in place, and left out of
+ * the ServerState and asked for again if this server keeps failing to serve the
+ * searches this phase reads the data with (issue #956). Nothing is logged here: this
+ * hook runs on every attempt in place, and the error is reported once, with the
+ * attempt which ended them, where a storage which failed to serve the operation is.
+ *
+ * @param csn the CSN of the change being replayed
+ * @param e the search which did not run
+ * @return the result which stops the operation
+ */
+ private SynchronizationProviderResult searchDidNotRun(CSN csn, SearchFailedException e)
+ {
+ return new SynchronizationProviderResult.StopProcessing(
+ ResultCode.UNAVAILABLE, e.report(csn, getBaseDN()));
+ }
+
+ /**
* Check that the broker associated to this ReplicationDomain has found
* a Replication Server and that this LDAP server is therefore able to
* process operations.
@@ -2730,15 +2786,28 @@
dependency = remotePendingChanges.checkDependencies(op, msg);
boolean replayDone = false;
boolean firstAttempt = true;
+ /*
+ * The search conflict resolution could not run on the last attempt, when it ended
+ * on one: the result code of the operation says nothing of it - it is the conflict
+ * the operation failed on - so the failure of the server below the loop is told
+ * here, and reported there once the attempts are spent.
+ */
+ SearchFailedException searchFailedResolvingConflict = null;
int retryCount = IN_PLACE_REPLAY_ATTEMPTS;
while (!dependency && !replayDone && retryCount-- > 0)
{
/*
- * The flag which says this domain is going down is read before the lock as well
- * as under it. The replay threads are a pool shared by every domain of this
- * server, so a thread which took a change of a domain which is going down should
- * not queue behind the wait for that domain: the changes of every other domain
- * are behind it in the same pool.
+ * What this attempt ends on, not what an earlier one did: an attempt the server
+ * refuses before it reaches the data makes no search, and the verdict below the
+ * loop reads the attempt which spent the last of them.
+ */
+ searchFailedResolvingConflict = null;
+ /*
+ * The flags which say this domain is going down, or that its data is being
+ * replaced, are read before the lock as well as under it. The replay threads are
+ * a pool shared by every domain of this server, so a thread which took a change
+ * of a domain which is going down should not queue behind the wait for that
+ * domain: the changes of every other domain are behind it in the same pool.
*
* The read under the lock is the one which decides; the one above it is a
* scheduling optimisation for the common case and nothing more. It cannot keep
@@ -2746,7 +2815,7 @@
* between the two reads, and a reader which arrives behind a queued writer blocks
* even on a lock which is not the fair kind.
*/
- boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled;
+ boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData;
if (!goingDown)
{
/*
@@ -2759,7 +2828,7 @@
replayReadLock.lock();
try
{
- goingDown = replayThreadShutdown.get() || shutdown.get() || disabled;
+ goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData;
if (!goingDown)
{
if (!firstAttempt)
@@ -2845,34 +2914,52 @@
else
{
ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO;
- if (op instanceof ModifyOperation)
+ try
{
- ModifyOperation castOp = (ModifyOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- ModifyMsg modifyMsg = (ModifyMsg) msg;
- resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg);
+ 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
}
- else if (op instanceof DeleteOperation)
+ catch (SearchFailedException e)
{
- DeleteOperation castOp = (DeleteOperation) op;
- dependency = remotePendingChanges.checkDependencies(castOp);
- resolution = dependency ? resolution : solveNamingConflict(castOp, msg);
+ /*
+ * Conflict resolution reads the data with a search of the entryUUID, and
+ * that search did not run: nothing was decided here, and whether the entry
+ * is still in the data is not known. Report the failure of the server it
+ * is rather than let a caller read "no entry" out of a search which never
+ * answered. It is logged once the attempts in place are spent rather than
+ * on every one of them, as a storage which failed to serve the operation
+ * is: a backend which is down for a while fails every attempt of every
+ * change delivered meanwhile.
+ */
+ searchFailedResolvingConflict = e;
+ resolution = ConflictResolution.SEARCH_FAILED;
}
- else if (op instanceof AddOperation)
- {
- AddOperation castOp = (AddOperation) op;
- AddMsg addMsg = (AddMsg) msg;
- dependency = remotePendingChanges.checkDependencies(castOp);
- resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg);
- }
- else if (op instanceof ModifyDNOperation)
- {
- ModifyDNOperation castOp = (ModifyDNOperation) op;
- ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg;
- dependency = remotePendingChanges.checkDependencies(modifyDNMsg);
- resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg);
- }
- // else: unknown type of operation ?! there is nothing to replay
if (!dependency)
{
@@ -2886,6 +2973,19 @@
recordChangeResolved(csn);
break;
+ case SEARCH_FAILED:
+ /*
+ * Conflict resolution could not read the data, so the change is not in
+ * it and nothing was concluded about the entry it targets. Give it the
+ * in-place attempts a storage which failed gets - a storage busy for a
+ * moment must not cost a session restart - and leave it out of the
+ * ServerState once they are spent, which the failure of the server
+ * below the loop reports and acts on. A change which is not in the data
+ * must not advance the ServerState (issue #889).
+ */
+ Thread.sleep(50);
+ break;
+
case FAILED:
if (serverErrorResultCode.equals(result))
{
@@ -2980,20 +3080,31 @@
* 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.
+ * loop below, however the attempts before it ended. So is an attempt whose
+ * conflict resolution could not read the data, which the attempt itself says
+ * rather than its result code: that one is the conflict the operation failed
+ * on, not the search which did not run (issue #956).
*/
if (isServerFailure(lastResult, serverErrorResultCode)
|| ResultCode.BUSY.equals(lastResult)
- || serverErrorResultCode.equals(lastResult))
+ || serverErrorResultCode.equals(lastResult)
+ || searchFailedResolvingConflict != null)
{
/*
* 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.
+ *
+ * The error reported is the operation's, unless the attempt ended on a search
+ * conflict resolution could not run: the operation then only says which conflict
+ * it failed on, and the search is what failed.
*/
+ final Object error = searchFailedResolvingConflict != null
+ ? searchFailedResolvingConflict.report(csn, getBaseDN())
+ : op.getErrorMessage();
final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get(
- op, csn, lastResult, op.getErrorMessage());
+ op, csn, lastResult, error);
logger.error(message);
replayErrorMsg = message.toString();
replayFailed = true;
@@ -3214,8 +3325,9 @@
*/
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.
+ // The ack has been published and the change is given back: the replication server
+ // delivers it again, now or - while a total update owns the session - after the
+ // import restarts it. There is nothing left to replay here.
return;
}
@@ -3524,13 +3636,19 @@
*/
remotePendingChanges.replayFailed(csn);
- if (shutdown.get() || disabled)
+ if (sessionHasAnOwner())
{
/*
* 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.
+ * thread and RSUpdater are already gone; restarting the one an import streams over
+ * would end the import on the entries which had arrived. The change given back here
+ * is forgotten with the rest of the pending changes when the ServerState is loaded
+ * again, from the backend or from the imported data, and the session started then
+ * asks for everything that state does not cover - or, when the total update it was
+ * given back for never begins, it is asked for by the next restart
+ * (see sessionHasAnOwner()).
*/
return true;
}
@@ -3642,9 +3760,10 @@
private void abandonReplay(CSN csn)
{
remotePendingChanges.replayFailed(csn);
- if (shutdown.get() || disabled)
+ if (sessionHasAnOwner())
{
- // The domain owns its session, and it forgets its pending changes on its way down.
+ // The domain, or the import into it, owns its session, and the pending changes are
+ // forgotten with the ServerState on its way down or at the end of the import.
return;
}
/*
@@ -3666,9 +3785,10 @@
final long stoppedSession;
synchronized (serviceStateLock)
{
- if (ownsItsSession())
+ if (sessionHasAnOwner())
{
- // The domain is going away or is being imported into: it owns its session.
+ // The domain is going away or is being imported into: the session is not this
+ // thread's to stop.
return;
}
disableService();
@@ -3694,6 +3814,11 @@
* import - in the meantime: the session this thread stopped is gone, so it has
* nothing left to start. Every stop and every start of a session is counted, so
* the generation alone tells one session from another.
+ *
+ * An import is not asked about here: the session it would stream over is the one
+ * this thread stopped, so none is streaming, and a total update which was asked
+ * for meanwhile needs the session started back to be answered at all - its
+ * request is what fails, and loudly, when no answer comes.
*/
return;
}
@@ -3782,24 +3907,79 @@
* @param uuid the Entry Unique ID.
* @return The current DN of the entry or null if there is no entry with
* the specified UUID.
+ * @throws SearchFailedException if the search did not run, so that whether there is
+ * such an entry is not known
*/
- private DN findEntryDN(String uuid)
+ private DN findEntryDN(String uuid) throws SearchFailedException
{
+ if (uuid == null)
+ {
+ // A change which carries no entryUUID names no entry.
+ return null;
+ }
+ /*
+ * The entryUUID comes off the wire and nothing validates it as one, so it is looked up
+ * as the value it is rather than read as part of a filter string: a value which
+ * carried a wildcard would be read as a substring filter, and one which did not parse
+ * - a dangling escape - would be a search which never runs, for as long as the change
+ * is asked for.
+ */
+ final SearchFilter filter = SearchFilter.createEqualityFilter(
+ getServerContext().getSchema().getAttributeType(ENTRYUUID_ATTRIBUTE_NAME),
+ ByteString.valueOfUtf8(uuid));
+ final InternalSearchOperation search =
+ conn.processSearch(newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, filter));
+ if (search.getResultCode() != ResultCode.SUCCESS)
+ {
+ if (search.getResultCode() == ResultCode.NO_SUCH_OBJECT && baseEntryIsAbsentFromALiveBackend())
+ {
+ /*
+ * The backend serves the base DN and has no base entry yet: the search ran, and
+ * nothing is below a base entry which is not there. This is the empty replica
+ * about to receive it.
+ */
+ return null;
+ }
+ /*
+ * The search did not run - the backend is offline or being rebuilt, or the storage
+ * failed to serve it - so it read nothing of the data: an entry it did not report
+ * is not an entry which is not there.
+ */
+ throw new SearchFailedException(uuid, search.getResultCode().getName() + " " + search.getErrorMessage());
+ }
+ final SearchResultEntry resultEntry = getFirstResult(search);
+ return resultEntry != null ? resultEntry.getName() : null;
+ }
+
+ /**
+ * Returns whether the backend which serves the base DN of this domain is there and
+ * holds no base entry, which is the state of an empty replica.
+ * <p>
+ * A search under the base DN of such a backend answers NO_SUCH_OBJECT, which is the
+ * answer of a backend which is offline or being rebuilt as well - and no backend
+ * answers SUCCESS with no entry for a base which is not there. The backend itself
+ * tells the two apart: the search ran over a backend which is there and empty, and it
+ * did not run over one which is gone or fails to answer.
+ *
+ * @return {@code true} if the backend is there and holds no base entry
+ */
+ private boolean baseEntryIsAbsentFromALiveBackend()
+ {
+ final LocalBackend<?> backend = getBackend();
+ if (backend == null)
+ {
+ // Nothing serves the base DN: the backend is offline or being rebuilt.
+ return false;
+ }
try
{
- final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid);
- InternalSearchOperation search = conn.processSearch(request);
- final SearchResultEntry resultEntry = getFirstResult(search);
- if (resultEntry != null)
- {
- return resultEntry.getName();
- }
+ return !backend.entryExists(getBaseDN());
}
catch (DirectoryException e)
{
- // never happens because the filter is always valid.
+ // The storage did not answer this any more than it answered the search.
+ return false;
}
- return null;
}
/** Outcome of the conflict resolution attempted after a replayed operation failed. */
@@ -3809,18 +3989,61 @@
REPLAY_AGAIN,
/** The change is already reflected in the data: there is nothing left to replay. */
NOTHING_TO_DO,
+ /**
+ * The search conflict resolution reads the data with did not run: nothing was
+ * decided, and the operation has to be attempted again.
+ */
+ SEARCH_FAILED,
/** The operation failed for a reason which is not a naming conflict. */
FAILED
}
/**
+ * Reports a search conflict resolution reads the data with and which did not run: it
+ * read nothing, so its result is no evidence about the data.
+ * <p>
+ * A search which failed and a search which found nothing look the same to a caller
+ * which only reads the entries it returned, and every caller here reads "no entry" as
+ * "the entry has been deleted": that answers {@link ConflictResolution#NOTHING_TO_DO},
+ * which records a change that was never applied as replayed, and the replication server
+ * never sends a change again which this replica reports itself past (issue #956).
+ */
+ private static final class SearchFailedException extends Exception
+ {
+ private static final long serialVersionUID = 1L;
+
+ /** The entryUUID the search which did not run was looking for. */
+ private final String entryUUID;
+
+ private SearchFailedException(String entryUUID, String cause)
+ {
+ super(cause);
+ this.entryUUID = entryUUID;
+ }
+
+ /**
+ * Describes the search which did not run for the change it was made for.
+ *
+ * @param csn the CSN of the change being replayed
+ * @param baseDN the base DN of the domain the change belongs to
+ * @return the error to report the change with
+ */
+ private LocalizableMessage report(CSN csn, DN baseDN)
+ {
+ return ERR_REPLAY_ENTRYUUID_SEARCH_FAILED.get(csn, baseDN, entryUUID, getMessage());
+ }
+ }
+
+ /**
* 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 the outcome of the conflict resolution
+ * @throws SearchFailedException if the data could not be read
*/
private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg)
+ throws SearchFailedException
{
ResultCode result = op.getResultCode();
ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT);
@@ -3905,8 +4128,10 @@
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
* @return the outcome of the conflict resolution
+ * @throws SearchFailedException if the data could not be read
*/
private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg)
+ throws SearchFailedException
{
ResultCode result = op.getResultCode();
DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT);
@@ -4403,11 +4628,12 @@
* Waits for the replay threads which are applying a change of this domain to be done
* with it.
* <p>
- * Called once {@link #disabled} or {@link #shutdown} has been set, which is what bounds
- * the wait: a replay thread reads those under {@link #replayReadLock}, the lock this
- * takes exclusively, so no attempt starts once this returns and what it waits for is the
- * attempts which were running already. The lock is released before returning for the
- * same reason - what keeps the replay out is the flag, not the lock.
+ * Called once {@link #disabled}, {@link #shutdown} or {@link #importingData} has been
+ * set, which is what bounds the wait: a replay thread reads those under
+ * {@link #replayReadLock}, the lock this takes exclusively, so no attempt starts once
+ * this returns and what it waits for is the attempts which were running already. The
+ * lock is released before returning for the same reason - what keeps the replay out is
+ * the flag, not the lock.
*/
private void awaitReplayDrained()
{
@@ -4936,6 +5162,16 @@
ImportExportContext ieCtx = getImportExportContext();
try
{
+ /*
+ * The replay of this domain is held off before the backend is taken away, the way
+ * disable() holds it off before the backend is taken away for an import run on this
+ * server: the flag keeps the attempts which have not started from starting, and the
+ * wait is for the ones which had. What this road can not do is stop the session -
+ * it is the session the entries are about to stream over, on this very thread.
+ */
+ importingData = true;
+ awaitReplayDrained();
+
if (!backend.supports(BackendOperation.LDIF_IMPORT))
{
ieCtx.setExceptionIfNoneSet(new DirectoryException(OTHER,
@@ -5001,6 +5237,25 @@
ResultCode.OTHER,
ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe))));
}
+ finally
+ {
+ /*
+ * The ServerState in memory is the one read from the data now, and the changes
+ * listed as pending must not outlive the one they went with, as they do not when
+ * the domain is disabled: a change given back while the import ran would stay
+ * listed, uncommitted and owned by nobody, and a commit stops at the first
+ * uncommitted change - the state would never move past it, and the replication
+ * server does not send a change again which the state it is given covers. Nothing
+ * listed a change meanwhile: the listener thread is the one running this import,
+ * and the replay threads gave up every attempt while the flag was set. The restart
+ * a replay thread may have asked for before the total update owned the session
+ * goes with them: the caller starts the session again from the reloaded state.
+ */
+ remotePendingChanges.clear();
+ sessionRestartRequested.set(false);
+ consecutiveSessionRestarts.set(0);
+ importingData = false;
+ }
}
if (ieCtx.getException() != null)
@@ -5276,6 +5531,29 @@
return shutdown.get() || disabled;
}
+ /**
+ * Whether the session of this domain has an owner other than the replay thread which
+ * would restart it after a failed replay: the domain itself, when it is shutting down or
+ * disabled ({@link #ownsItsSession()}), or a total update into this replica.
+ * <p>
+ * The total update owns the session from the moment it is asked for, not from the
+ * moment its entries stream: the {@code InitializeTargetMsg} which answers the request
+ * arrives over that session, so a restart made while it is on its way loses it, and the
+ * import which follows reads its entries over the same session - stopping it ends the
+ * import on the entries which had arrived. The import starts the next session itself,
+ * from the state it loaded. A change given back while the total update owned the session
+ * is not asked for again by anyone until then; if no import follows - the request was
+ * refused, or gave up waiting - it stays listed until the next failed replay restarts
+ * the session, which has the replication server send it again with everything after it.
+ * Listed, it holds the ServerState back as well: a commit moves the state no further than
+ * the oldest uncommitted change, so the state in memory, and the one persisted from it,
+ * stop at the change until that restart.
+ */
+ private boolean sessionHasAnOwner()
+ {
+ return ownsItsSession() || importInProgress();
+ }
+
@Override
protected void restartService()
{
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 fe6c496..f839681 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -686,3 +686,7 @@
the domain is shutting down, or it is disabled for the length of a total update. The change is \
stored and takes effect when the session is started again, which a domain left disabled by a \
failed import or restore never does
+ERR_REPLAY_ENTRYUUID_SEARCH_FAILED_322=Could not read the data to check change %s for a conflict \
+ in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is not \
+ applied on what a search which read nothing seemed to say about the data, and is not recorded \
+ as replayed
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 f61b23d..786fce9 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
@@ -644,16 +644,18 @@
// Check for registered short circuits.
final String key = keyFor(operation.getOperationType(), section);
Integer resultCode = shortCircuits.get(key);
- if (resultCode != null)
+ if (resultCode != null && appliesTo(key, operation))
{
final int reached = shortCircuitCounts.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet();
+ final int letThroughFirst = shortCircuitSkips.getOrDefault(key, 0);
final Integer maxTimes = shortCircuitLimits.get(key);
- if (maxTimes == null || reached <= maxTimes)
+ if (reached > letThroughFirst && (maxTimes == null || reached <= letThroughFirst + 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.
+ // The operations before the short circuit are let through, and so are the ones after
+ // it was applied as many times as it was asked for: this is how a transient failure
+ // which starts, or ends, part way through a sequence of operations is simulated.
}
/*
@@ -740,6 +742,23 @@
/** How many times a registered short circuit must be applied, when it is limited. */
private static final Map<String, Integer> shortCircuitLimits = new ConcurrentHashMap<>();
+ /** How many operations a registered short circuit lets through before it applies. */
+ private static final Map<String, Integer> shortCircuitSkips = new ConcurrentHashMap<>();
+
+ /**
+ * Which operations a registered short circuit is for, when it is not for every operation
+ * of its type: the ones it is not for are neither short circuited nor counted.
+ */
+ private static final Map<String, Predicate<PluginOperation>> shortCircuitFilters =
+ new ConcurrentHashMap<>();
+
+ /** Returns whether the short circuit registered under the given key is for the given operation. */
+ private static boolean appliesTo(String key, PluginOperation operation)
+ {
+ final Predicate<PluginOperation> filter = shortCircuitFilters.get(key);
+ return filter == null || filter.test(operation);
+ }
+
/**
* 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
@@ -765,10 +784,13 @@
public static void registerShortCircuit(OperationType operation, String section, int resultCode)
{
final String key = keyFor(operation, section);
- // 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.
+ // This registration applies to every operation, and it counts from zero: a limit, a
+ // number of operations let through or a count left behind by a previous registration
+ // is not part of it.
shortCircuitCounts.remove(key);
shortCircuitLimits.remove(key);
+ shortCircuitSkips.remove(key);
+ shortCircuitFilters.remove(key);
shortCircuits.put(key, resultCode);
}
@@ -783,13 +805,62 @@
*/
public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes)
{
+ registerShortCircuit(operation, section, resultCode, 0, maxTimes);
+ }
+
+ /**
+ * Register a short circuit which lets the given number of operations through before it
+ * applies, then applies to the given number of operations, the ones which follow being
+ * let through again: this is how a transient failure which starts part way through a
+ * sequence of operations is simulated - the second search of an attempt failing while
+ * the first one ran, say.
+ *
+ * @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 letThroughFirst How many operations must be let through before the short
+ * circuit applies.
+ * @param maxTimes How many operations must be short circuited after them.
+ */
+ public static void registerShortCircuit(OperationType operation, String section, int resultCode,
+ int letThroughFirst, int maxTimes)
+ {
final String key = keyFor(operation, section);
shortCircuitCounts.remove(key);
+ shortCircuitFilters.remove(key);
+ shortCircuitSkips.put(key, letThroughFirst);
shortCircuitLimits.put(key, maxTimes);
shortCircuits.put(key, resultCode);
}
/**
+ * Register a short circuit like
+ * {@link #registerShortCircuit(OperationType, String, int, int, int)}, for some of the
+ * operations of the given type only: the ones the predicate does not accept are let
+ * through without being counted, as if the short circuit were not there.
+ * <p>
+ * The operations of one type which reach a plugin point are not all the test's: the
+ * server makes internal operations of its own on its own schedule - the ServerState flush
+ * thread of a replication domain writes the base entry with a Modify on its tick, say -
+ * and a short circuit which counts them takes a let-through, or a refusal, meant for the
+ * operation the test is driving. A test which counts its operations one by one names them.
+ *
+ * @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 letThroughFirst How many of the operations the predicate accepts must be let
+ * through before the short circuit applies.
+ * @param maxTimes How many of them must be short circuited after those.
+ * @param appliesTo Which operations of that type the short circuit is for.
+ */
+ public static void registerShortCircuit(OperationType operation, String section, int resultCode,
+ int letThroughFirst, int maxTimes, Predicate<PluginOperation> appliesTo)
+ {
+ registerShortCircuit(operation, section, resultCode, letThroughFirst, maxTimes);
+ shortCircuitFilters.put(keyFor(operation, section), appliesTo);
+ }
+
+ /**
* Deregister a short circuit for the given operation type and plugin point.
* @param operation The type of operation the short circuit applies to.
* @param section The plugin point the short circuit applies to.
@@ -799,6 +870,8 @@
final String key = keyFor(operation, section);
shortCircuits.remove(key);
shortCircuitLimits.remove(key);
+ shortCircuitSkips.remove(key);
+ shortCircuitFilters.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/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
index cb74306..29659d1 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
@@ -18,11 +18,14 @@
package org.opends.server.replication.plugin;
import static org.assertj.core.api.Assertions.*;
+import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.testng.Assert.*;
+import java.util.ArrayList;
+import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -33,6 +36,8 @@
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ModifyDNOperation;
+import org.opends.server.core.ModifyOperationBasis;
+import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
@@ -40,8 +45,11 @@
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
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.UpdateMsg;
import org.opends.server.types.Entry;
+import org.opends.server.types.OperationType;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -52,6 +60,10 @@
{
private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false);
+ /** The monitor attributes which count the naming conflicts a domain solved, and did not. */
+ private static final String RESOLVED_NAMING_CONFLICTS = "resolved-naming-conflicts";
+ private static final String UNRESOLVED_NAMING_CONFLICTS = "unresolved-naming-conflicts";
+
private DN baseDN;
private LDAPReplicationDomain domain;
private CSNGenerator gen;
@@ -245,6 +257,531 @@
}
/**
+ * Test case for [Issue 956]: conflict resolution reads the data with a search of the
+ * entryUUID, and a search which did not run is no evidence about the data - the entry
+ * it did not report is not an entry which was deleted.
+ * <p>
+ * The change replayed here was made on the master under a DN this replica does not
+ * have: the entry lives under another one, the way it does after a rename which was
+ * replayed first, and only the entryUUID search finds it. So the change is applied
+ * only if that search is given another chance once the storage serves it again.
+ * Reading its failure as "the entry has been deleted" answers NOTHING_TO_DO, which
+ * records the change as replayed and loses it for good - the replication server never
+ * sends a change this replica reports itself past.
+ */
+ @Test
+ public void modifyIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final String phoneNumber = "01 02 45";
+
+ // The DN the change carries is the one the entry had on the master. Here the entry
+ // is the one which was just added, which only the entryUUID search finds.
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ /*
+ * The storage does not serve the search for the first attempts and serves it after
+ * them: a failure which lasts less than the attempts made in place, the way a
+ * backend which is being rebuilt or a connection which was lost does. The short
+ * circuit is put in force right before the replay and dropped right after it - it
+ * applies to every search of this server while it is registered, and the replay
+ * here runs on this thread.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", phoneNumber), entryUUID));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ final Entry replayedEntry = DirectoryServer.getEntry(entry.getName());
+ assertEquals(replayedEntry.parseAttribute("telephonenumber").asString(), phoneNumber,
+ "the change was not applied: a search which could not run was read as a deleted entry");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the entryUUID searches which check a replayed Add for a
+ * conflict read the data the same way, and a search which did not run is no evidence
+ * about it either. The first of them checks whether the Add was replayed here already.
+ * <p>
+ * The Add is delivered a second time - the replication server sends again what a
+ * replica does not report itself past - and the entry was renamed here since it was
+ * added: only the entryUUID search finds it. An entry that search did not report is
+ * not an entry which is not there: reading it that way adds the entry a second time,
+ * under its former DN, and the data holds one entryUUID twice.
+ */
+ @Test
+ public void addIsNotReplayedTwiceWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("addWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final RDN renamedRDN = RDN.valueOf("cn=renamedAfterTheAdd");
+ final ModifyDNOperation rename =
+ getRootConnection().processModifyDN(entry.getName(), renamedRDN, true);
+ assertEquals(rename.getResultCode(), ResultCode.SUCCESS);
+ final CSN csn = gen.newCSN();
+
+ // The first attempt fails its first search; the second attempt has it served.
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1);
+ try
+ {
+ replayMsg(addMsg(entry, csn, getEntryUUID(baseDN), entryUUID));
+ assertShortCircuitSpentBy(1);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(entryExists(entry.getName()),
+ "the entry was added a second time: a search which could not run was read as an "
+ + "Add which was not replayed here yet");
+ assertTrue(entryExists(baseDN.child(renamedRDN)), "the renamed entry is gone");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which is in the data must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the search which checks that the parent of a replayed
+ * Add is still the one the change was made under fails, and the one before it - the
+ * check that the Add was not replayed here already - ran.
+ * <p>
+ * A parent that search did not report is not a parent which was deleted: the Add is
+ * attempted again once the storage serves the search, and no naming conflict is
+ * counted for a search which read nothing. Reading it as a parent which is gone hands
+ * the Add to conflict resolution as the naming conflict it is not - and renames the
+ * entry under the base DN as a conflicting entry, a divergence which is left for an
+ * administrator to repair by hand, when the search conflict resolution makes fails as
+ * well.
+ */
+ @Test
+ public void addIsRetriedWhileTheParentEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry parent = addParentEntry("addWhoseParentSearchCanNotRun");
+ final String parentUUID = getEntryUUID(parent.getName());
+ final Entry child = makeChildEntry("addedWhileTheParentSearchFailed", parent.getName());
+ final CSN csn = gen.newCSN();
+ final long resolvedConflicts = getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS);
+ final long unresolvedConflicts = getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS);
+
+ /*
+ * The first search of the first attempt - the check for an Add replayed already - is
+ * let through, the parent check right after it fails, and every search after that
+ * one is served. A parent search read as a parent which is gone hands the Add to
+ * conflict resolution, whose own search finds the parent where it was and rewrites
+ * the message to the DN the Add already carries: the entry lands where it should
+ * either way, and what tells the two apart is the naming conflict counted for a
+ * search which read nothing.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1, 1);
+ try
+ {
+ replayMsg(addMsg(child, csn, parentUUID, "1c3c2c4d-2b5e-4b9f-8a7c-3d5e6f7a8b9c"));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(child.getName()),
+ "the entry was not added under its parent: a parent search which could not run "
+ + "was read as a parent which is gone");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ assertEquals(getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS), resolvedConflicts,
+ "a search which did not run is not a naming conflict which was resolved");
+ assertEquals(getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS), unresolvedConflicts,
+ "a search which did not run is not a naming conflict which could not be resolved");
+ }
+
+ /**
+ * Test case for [Issue 956]: the search conflict resolution reads the data with once a
+ * replayed Add failed on a genuine conflict fails, and the checks before the Add ran.
+ * <p>
+ * The parent of the entry was renamed here, so the Add carries a DN which is not
+ * where the parent is anymore: a conflict which is solved by adding the entry under
+ * the parent's current DN, once the search which finds that DN runs. A search which
+ * did not run is not a parent which is gone, and the conflict is counted once, when
+ * it is solved - not for the search which read nothing.
+ */
+ @Test
+ public void addIsRetriedWhileTheConflictResolutionSearchCanNotRun() throws Exception
+ {
+ final Entry parent = addParentEntry("addWhoseConflictSearchCanNotRun");
+ final String parentUUID = getEntryUUID(parent.getName());
+ final Entry child = makeChildEntry("addedWhileTheConflictSearchFailed", parent.getName());
+ final String entryUUID = "2d4d3d5e-3c6f-4ca0-9b8d-4e6f7a8b9cad";
+ final CSN csn = gen.newCSN();
+
+ final RDN renamedParentRDN = RDN.valueOf("ou=renamedBeforeTheAdd");
+ final ModifyDNOperation renameParent =
+ getRootConnection().processModifyDN(parent.getName(), renamedParentRDN, true);
+ assertEquals(renameParent.getResultCode(), ResultCode.SUCCESS);
+ final DN expectedDN = baseDN.child(renamedParentRDN).child(child.getName().rdn());
+ final long resolvedConflicts = getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS);
+ final long unresolvedConflicts = getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS);
+
+ /*
+ * The two searches which check the Add before it runs are let through - they find
+ * the parent under its new DN, which is what fails the Add on a conflict - and the
+ * search conflict resolution then reads the data with is the one which fails. The
+ * next attempt has every search served.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2, 1);
+ try
+ {
+ replayMsg(addMsg(child, csn, parentUUID, entryUUID));
+ assertShortCircuitSpentBy(3);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(expectedDN),
+ "the entry was not added under the current DN of its parent: a search which could "
+ + "not run was read as a parent which is gone");
+ // A parent read as gone puts the entry under the base DN as a conflicting entry, with
+ // its entryUUID added to its RDN.
+ assertFalse(entryExists(DN.valueOf(
+ "entryuuid=" + entryUUID + "+" + child.getName().rdn() + "," + TEST_ROOT_DN_STRING)),
+ "the entry was renamed under the base DN as a conflicting entry");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ assertEquals(getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS), resolvedConflicts + 1,
+ "the renamed parent is one naming conflict, solved once the search ran");
+ assertEquals(getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS), unresolvedConflicts,
+ "a search which did not run is not a naming conflict which could not be resolved");
+ }
+
+ /**
+ * Test case for [Issue 956]: a replayed Delete reads the data with the same search
+ * once it failed on a conflict, and rides on the same retry.
+ * <p>
+ * The entry was renamed here, so the Delete carries a DN which is not the entry's
+ * anymore, and only the entryUUID search finds it. Reading the search's failure as an
+ * entry which was deleted already answers NOTHING_TO_DO: the entry stays, and the
+ * change is recorded as replayed.
+ */
+ @Test
+ public void deleteIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("deleteWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new DeleteMsg(staleDN, csn, entryUUID));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(entryExists(entry.getName()),
+ "the entry was not deleted: a search which could not run was read as an entry "
+ + "which was deleted already");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: a replayed Modify DN reads the data with the same search
+ * once it failed on a conflict, and rides on the same retry.
+ * <p>
+ * {@code solveNamingConflict(ModifyDNOperation)} declares {@code throws Exception}
+ * rather than the search failure alone, so this is the one path where the failure
+ * reaches the replay loop through a declaration which does not name it.
+ */
+ @Test
+ public void modifyDnIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyDnWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final RDN newRDN = RDN.valueOf("cn=renamedWhileTheSearchFailed");
+ final CSN csn = gen.newCSN();
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new ModifyDNMsg(staleDN, csn, entryUUID, null, false, null, newRDN.toString()));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(baseDN.child(newRDN)),
+ "the entry was not renamed: a search which could not run was read as an entry "
+ + "which is not in the data anymore");
+ assertFalse(entryExists(entry.getName()), "the entry kept its former DN");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956], the half which closes [Issue 889] for this path: a
+ * change whose entryUUID search never runs is left out of the ServerState once the
+ * attempts in place are spent, so that the replication server sends it again.
+ * <p>
+ * The result code of every attempt is the conflict the operation failed on, which the
+ * exhaustion exit does not read as a failure of the server: without the attempt itself
+ * telling that its search did not run, the exit reads the attempts as conflict
+ * resolution rewriting an operation which keeps failing, and skips the change - the
+ * CSN is committed, and a change which is not in the data is never asked for again.
+ */
+ @Test
+ public void modifyIsLeftOutOfTheServerStateWhenTheEntryUUIDSearchNeverRuns() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseSearchNeverRuns");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ // No number of times: every attempt in place fails its search.
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have made its search");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(domain.getServerState().cover(csn),
+ "a change whose search never ran is not in the data and must not advance the ServerState");
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("telephonenumber"))
+ .as("the change was applied to the entry the searches never found").isEmpty();
+ /*
+ * The result code of the last attempt is the conflict the operation failed on, which
+ * says nothing of the search: the line which reports the change must carry the search
+ * which did not run, which is the one thing that names the entryUUID it was made for.
+ */
+ final List<String> reports = exhaustionExitRecordsOf(csn);
+ assertThat(reports).as("the change was not reported once the attempts in place were spent")
+ .isNotEmpty();
+ assertThat(reports)
+ .as("the exhaustion exit reports the error of the operation, not the search which did not run")
+ .allMatch(record -> record.contains(entryUUID));
+ }
+
+ /**
+ * Test case for [Issue 956]: the exhaustion exit reports the attempt which spent the
+ * last of the attempts in place, not an earlier one which ended on a search conflict
+ * resolution could not run.
+ * <p>
+ * The first attempt reaches the data, fails on the conflict, and its search does not
+ * run; the server then refuses every attempt after it before the data is reached. Each
+ * of these is a failure of the server, and the change is left out of the ServerState
+ * either way - what the line which reports it carries is the question. Its result
+ * code is the last attempt's, and so must be the error next to it: a line which reads
+ * the server refusing the operation next to a search which did not run names two
+ * causes, and the operator chases the wrong one.
+ */
+ @Test
+ public void theExhaustionExitReportsTheAttemptWhichSpentTheLastOfThem() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseLastAttemptIsRefused");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ /*
+ * The first attempt is let through to the data and fails on the conflict of the
+ * stale DN; its search is the one the short circuit stops. The attempts after it are
+ * refused before they reach the data, the way a backend which went offline after the
+ * first attempt refuses them: no search is made on any of them.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1);
+ // The replayed operation only, named by its CSN: see the flush below.
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse",
+ ResultCode.UNAVAILABLE.intValue(), 1, LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS - 1,
+ op -> csn.equals(OperationContext.getCSN(op)));
+ try
+ {
+ /*
+ * The ServerState flush thread saves the state with a Modify of the base entry on
+ * its tick, which the add above made dirty: a tick which lands among the attempts in
+ * place is a Modify the short circuit meets like any other. Made here rather than
+ * left to the tick, so that the case says what it does about it every time instead
+ * of once in a while: that Modify is not the replayed operation, and the short
+ * circuit must neither let it through in place of the first attempt nor refuse it
+ * in place of a later one.
+ */
+ flushLikeTheStateFlushThread();
+ assertEquals(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse"), 0,
+ "the Modify of the state flush thread was counted against the short circuit");
+
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") >= 1,
+ "the first attempt must have made its search");
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have been made");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(domain.getServerState().cover(csn),
+ "a change the server kept refusing is not in the data and must not advance the ServerState");
+ final List<String> reports = exhaustionExitRecordsOf(csn);
+ assertThat(reports).as("the change was not reported once the attempts in place were spent")
+ .isNotEmpty();
+ // The last attempt was refused before it reached the data and made no search: a line
+ // which names the entryUUID reports the search of an earlier attempt next to its result.
+ assertThat(reports).as("the exhaustion exit reports an attempt other than the last one")
+ .allMatch(record -> !record.contains(entryUUID));
+ }
+
+ /**
+ * Test case for [Issue 956]: the base entry of the domain replayed into a replica
+ * which has none.
+ * <p>
+ * Two empty replicas share the generation ID of an empty backend, so no initialization
+ * is needed and the first change replayed is the base entry itself. The searches which
+ * check that Add for a conflict run under the base DN, which the backend serves and
+ * has no entry for: they answer NO_SUCH_OBJECT, which is the answer of a backend which
+ * is offline as well. Here the search ran and nothing is below a base entry which is
+ * not there, so the Add must go through rather than be retried until the give-up
+ * budget skips it as a change this replica can not apply.
+ */
+ @Test
+ public void baseEntryIsAddedToAnEmptyReplica() throws Exception
+ {
+ // The backend, without its base entry.
+ TestCaseUtils.initializeTestBackend(false);
+ final Entry base = TestCaseUtils.makeEntry(
+ "dn: " + TEST_ROOT_DN_STRING,
+ "objectClass: top",
+ "objectClass: organization",
+ "o: test");
+ final CSN csn = gen.newCSN();
+
+ replayMsg(addMsg(base, csn, null, "7c1a0d2e-4b6f-4c8a-9e1d-3f5b7a9c1e2d"));
+
+ assertTrue(entryExists(base.getName()),
+ "the base entry of an empty replica must land: its searches found nothing, they did not fail");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the entryUUID a change carries is looked up as a value,
+ * not read as a filter.
+ * <p>
+ * The entryUUID comes off the wire and nothing validates it as one. Built into a
+ * filter string, a value which does not parse as a filter is a search which never
+ * runs - a permanent condition retried as a transient one, for as long as the change
+ * is asked for, until the give-up budget skips it and raises an alert. Looked up as a
+ * value, such an entryUUID names no entry, which is what a search which ran and found
+ * nothing says: the change is resolved as one on an entry which is not in the data.
+ */
+ @Test
+ public void anEntryUUIDWhichIsNotOneNamesNoEntry() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseEntryUUIDIsNotOne");
+ // A value no filter string parses: the backslash escapes nothing.
+ final String entryUUID = getEntryUUID(entry.getName()) + "\\";
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+
+ assertTrue(domain.getServerState().cover(csn),
+ "a change on an entry which is not in the data is a conflict which is resolved, and recorded");
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("telephonenumber"))
+ .as("a change on an entryUUID which is not one was applied to an entry").isEmpty();
+ }
+
+ /**
+ * The records of the error log which report the provided change once its attempts in
+ * place were spent. The record the error logger writes carries the id of the message
+ * rather than its text.
+ */
+ private static List<String> exhaustionExitRecordsOf(CSN csn)
+ {
+ final String exhaustionExit = "msgID=" + ERR_ERROR_REPLAYING_OPERATION.ordinal();
+ final List<String> reports = new ArrayList<>();
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ if (record.contains(exhaustionExit) && record.contains(csn.toString()))
+ {
+ reports.add(record);
+ }
+ }
+ return reports;
+ }
+
+ /**
+ * Asserts that the searches a short circuit was registered over were made - the ones
+ * let through before it and the ones it applied to - and that the search after them
+ * was made as well.
+ * <p>
+ * The count includes the searches let through once the short circuit was spent, so a
+ * count past what it was registered over says that the budget was used and that the
+ * search after it ran. Read before the short circuit is deregistered, which drops the
+ * count with it.
+ *
+ * @param searchesRegisteredOver the searches let through before the short circuit plus
+ * the ones it applied to
+ */
+ private void assertShortCircuitSpentBy(int searchesRegisteredOver)
+ {
+ assertTrue(
+ ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") > searchesRegisteredOver,
+ "the short circuit must have been spent by the attempts in place");
+ }
+
+ private Entry addParentEntry(String ou) throws Exception
+ {
+ return TestCaseUtils.addEntry(
+ "dn: ou=" + ou + "," + TEST_ROOT_DN_STRING,
+ "objectClass: top",
+ "objectClass: organizationalUnit",
+ "ou: " + ou);
+ }
+
+ private Entry makeChildEntry(String cn, DN parentDN) throws Exception
+ {
+ return TestCaseUtils.makeEntry(
+ "dn: cn=" + cn + "," + parentDN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: " + cn,
+ "sn: Amar");
+ }
+
+ /**
* Test that when a previous conflict is resolved because
* a delete operation has removed one of the conflicting entries
* the other conflicting entry is correctly renamed to its original name.
@@ -317,6 +854,24 @@
assertThat(resultEntry.getAllAttributes(LDAPReplicationDomain.DS_SYNC_CONFLICT)).isEmpty();
}
+ /**
+ * Makes the Modify the ServerState flush thread makes on its tick: an internal
+ * synchronization Modify of the base entry, which is not synchronized itself. The
+ * attribute is a harmless one rather than ds-sync-state, which is the flush thread's to
+ * write.
+ */
+ private void flushLikeTheStateFlushThread()
+ {
+ final ModifyOperationBasis op = new ModifyOperationBasis(getRootConnection(),
+ nextOperationID(), nextMessageID(), null,
+ baseDN, generatemods("description", "written on the tick of the flush thread"));
+ op.setInternalOperation(true);
+ op.setSynchronizationOperation(true);
+ op.setDontSynchronize(true);
+ op.run();
+ assertEquals(op.getResultCode(), ResultCode.SUCCESS, op.getErrorMessage().toString());
+ }
+
private Entry createAndAddEntry(String commonName) throws Exception
{
// @formatter:off
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
new file mode 100644
index 0000000..3bedecb
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -0,0 +1,385 @@
+/*
+ * 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 java.nio.charset.StandardCharsets.*;
+import static org.assertj.core.api.Assertions.*;
+import static org.opends.messages.ReplicationMessages.*;
+import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.core.DirectoryServer.*;
+import static org.testng.Assert.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ResultCode;
+import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.core.DirectoryServer;
+import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.protocol.DoneMsg;
+import org.opends.server.replication.protocol.EntryMsg;
+import org.opends.server.replication.protocol.InitializeRequestMsg;
+import org.opends.server.replication.protocol.InitializeTargetMsg;
+import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.replication.protocol.UpdateMsg;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.types.Entry;
+import org.opends.server.types.OperationType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Tests the replay of a change while this replica is the target of a total update.
+ * <p>
+ * The import of a total update streams over the session of the domain, on its listener
+ * thread, and the backend it replaces is deregistered for the length of it. A change which
+ * was queued for replay before the {@code InitializeTargetMsg} arrived is replayed into no
+ * backend: whatever such a replay decides is about to be overwritten by the import, and the
+ * one thing it must not do is stop the session the import is reading (issue #956). The same
+ * holds from the moment the total update is asked for: the answer to the request arrives
+ * over that session, so a replay which fails while it is on its way must not restart it.
+ * <p>
+ * The exporter is a broker of this test, so that the test says when the entries arrive: the
+ * change is replayed while the import is waiting for them - or, for the request, while the
+ * exporter is holding the answer.
+ */
+@SuppressWarnings("javadoc")
+public class ReplayDuringImportTest extends ReplicationTestCase
+{
+ /**
+ * The memory backend of {@code o=test} loses its data when it is disabled and enabled
+ * back, which is what an import does to the backend it replaces: a total update needs a
+ * backend which keeps what was imported into it.
+ */
+ private static final String EXAMPLE_DN = "dc=example,dc=com";
+ private static final int RS_ID = 611;
+ private static final int DS_ID = 1;
+ private static final int EXPORTER_ID = 2;
+ private static final int INIT_WINDOW = 100;
+ private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false);
+ /** An entry of the exporter's data, and its entryUUID. */
+ private static final String IMPORTED_ENTRY_DN = "cn=imported,ou=People," + EXAMPLE_DN;
+ private static final String IMPORTED_ENTRY_UUID = "21111111-1111-1111-1111-111111111113";
+
+ private DN baseDN;
+ private ReplicationServer replicationServer;
+ private LDAPReplicationDomain domain;
+ private TestSynchronousReplayQueue queue;
+ private ReplicationBroker exporter;
+ private CSNGenerator gen;
+
+ @BeforeMethod
+ public void setUpLocal() throws Exception
+ {
+ baseDN = DN.valueOf(EXAMPLE_DN);
+ TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN);
+
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ rsPort, "replayDuringImportTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
+
+ final SortedSet<String> replServers = new TreeSet<>();
+ replServers.add("localhost:" + rsPort);
+ final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID, replServers);
+ conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);
+ queue = new TestSynchronousReplayQueue();
+ domain = MultimasterReplication.createNewDomain(conf, queue);
+ domain.start();
+ assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
+
+ exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000);
+ gen = new CSNGenerator(201, 0);
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception
+ {
+ try
+ {
+ stop(exporter);
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+ finally
+ {
+ remove(replicationServer);
+ }
+ }
+
+ /**
+ * A change replayed while the import streams must leave the session to the import.
+ * <p>
+ * The change is given back at the top of its first attempt: the data it would be applied
+ * to is being replaced, so nothing is attempted into the backend the import took away,
+ * nothing is reported, and the session is left to the import - which streams every entry
+ * to its end. Without the hold-off the operation is refused with NO_SUCH_OBJECT - nothing
+ * serves the base DN - and the entryUUID search conflict resolution reads the data with
+ * can not run either: the attempts in place are spent into no backend and the exit
+ * reports the change; without the owner the total update is, the session is then
+ * restarted for the change to be delivered again, which stops the broker the import is
+ * reading, and the import ends on the entries which had arrived with nothing to say it.
+ */
+ @Test(timeOut = 120_000)
+ public void aReplayDuringTheImportLeavesTheSessionToTheImport() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+
+ // Queued before the InitializeTargetMsg arrived, replayed into no backend.
+ final CSN csn = gen.newCSN();
+ replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed during the import"), entryUUID));
+
+ finishImport(exported);
+
+ for (String ldif : exported)
+ {
+ final DN dn = dnOf(ldif);
+ assertTrue(entryExists(dn), "the import ended before " + dn
+ + " arrived: the session it streams over was stopped from under it");
+ }
+ /*
+ * The two roads which leave the session to the import are told apart here: the
+ * hold-off gives the change back before an attempt is made, the guard on the restart
+ * after the attempts are spent. The exhaustion exit is the one thing the first road
+ * leaves no record of.
+ */
+ assertThat(errorLogRecordsOf(ERR_ERROR_REPLAYING_OPERATION.ordinal(), csn))
+ .as("the change was attempted into no backend instead of being given back at once")
+ .isEmpty();
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was asked for again, which restarts the session the import streams over")
+ .isEmpty();
+ }
+
+ /**
+ * A change given back while the import ran must not hold the ServerState back once the
+ * import has replaced the data.
+ * <p>
+ * A change which is given back stays listed as pending and uncommitted - that is what
+ * has the replication server send it again - and a commit advances the ServerState no
+ * further than the oldest uncommitted change. The state the import loads is the
+ * exporter's, which covers the change already, so nothing sends it again: left listed,
+ * it would stop the ServerState of this replica for good.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeGivenBackDuringTheImportDoesNotHoldTheServerStateBack() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ replayMsg(new ModifyMsg(gen.newCSN(), DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed during the import"), entryUUID));
+ finishImport(exported);
+
+ // A change on an entry the import brought, replayed once the import is over.
+ final DN importedDN = DN.valueOf(IMPORTED_ENTRY_DN);
+ final CSN csn = gen.newCSN();
+ replayMsg(new ModifyMsg(csn, importedDN,
+ generatemods("description", "replayed after the import"), IMPORTED_ENTRY_UUID));
+
+ assertThat(DirectoryServer.getEntry(importedDN).getAllAttributes("description"))
+ .as("a change replayed after the import was not applied").isNotEmpty();
+ assertTrue(domain.getServerState().cover(csn), "a change applied after the import was not"
+ + " recorded: the change given back during the import is still listed and holds the"
+ + " ServerState back");
+ }
+
+ /**
+ * A total update this replica asked for owns the session from the request, not from the
+ * first entry: the {@code InitializeTargetMsg} which answers the request arrives over
+ * that session, and a restart made while the answer is on its way loses it.
+ * <p>
+ * The backend is live for the length of the request - nothing has been taken away yet -
+ * so the change is attempted, every attempt ends on an entryUUID search which does not
+ * run, and the exhaustion exit reports it: what is refused is the restart which would
+ * have followed, and the retry warning which goes with it. The exporter then answers the
+ * request, and its entries stream to their end over the session which was left alone.
+ */
+ @Test(timeOut = 120_000)
+ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final String[] exported = exportedEntries();
+
+ // The request is out, and the exporter holds it until the change below has been replayed.
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed while the request was on its way"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have made its search: the backend is live while the"
+ + " request is on its way, so nothing holds the replay off");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertThat(errorLogRecordsOf(ERR_ERROR_REPLAYING_OPERATION.ordinal(), csn))
+ .as("the attempts in place were spent, which the exhaustion exit reports").isNotEmpty();
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was asked for again, which restarts the session the answer to the"
+ + " request arrives over")
+ .isEmpty();
+ assertTrue(domain.isConnected(), "the session the request was made over was stopped");
+
+ answerImportRequest(exported.length);
+ finishImport(exported);
+ for (String ldif : exported)
+ {
+ final DN dn = dnOf(ldif);
+ assertTrue(entryExists(dn), "the import ended before " + dn
+ + " arrived: the answer to the request was lost with the session it was made over");
+ }
+ }
+
+ /**
+ * Has the exporter start a total update into this replica, and returns once the backend
+ * of the domain is deregistered for it: from then on the import is reading the session,
+ * and a change replayed here is replayed into no backend.
+ */
+ private void startImportInto(int entryCount) throws Exception
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
+ final long deadline = System.currentTimeMillis() + 30_000;
+ while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
+ {
+ assertTrue(System.currentTimeMillis() < deadline,
+ "the import did not deregister the backend of the domain");
+ Thread.sleep(20);
+ }
+ }
+
+ /**
+ * Has the exporter answer the total update this replica asked for: the requestor of the
+ * {@code InitializeTargetMsg} is this replica, so the import runs in the context the
+ * request acquired.
+ */
+ private void answerImportRequest(int entryCount) throws Exception
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, DS_ID, entryCount, INIT_WINDOW));
+ }
+
+ /** Has the exporter send the entries of the total update, and waits for the import to end. */
+ private void finishImport(String... ldifEntries) throws Exception
+ {
+ int msgId = 0;
+ for (String ldif : ldifEntries)
+ {
+ exporter.publish(new EntryMsg(EXPORTER_ID, DS_ID, ldif.getBytes(UTF_8), ++msgId));
+ }
+ exporter.publish(new DoneMsg(EXPORTER_ID, DS_ID));
+ final long deadline = System.currentTimeMillis() + 60_000;
+ while (domain.ieRunning())
+ {
+ assertTrue(System.currentTimeMillis() < deadline, "the import did not end");
+ Thread.sleep(50);
+ }
+ }
+
+ /** The data of the exporter: the base entry and two entries below it. */
+ private static String[] exportedEntries()
+ {
+ return new String[] {
+ "dn: " + EXAMPLE_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: domain\n"
+ + "dc: example\n"
+ + "entryUUID: 21111111-1111-1111-1111-111111111111\n"
+ + "\n",
+ "dn: ou=People," + EXAMPLE_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: organizationalUnit\n"
+ + "ou: People\n"
+ + "entryUUID: 21111111-1111-1111-1111-111111111112\n"
+ + "\n",
+ "dn: " + IMPORTED_ENTRY_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: person\n"
+ + "cn: imported\n"
+ + "sn: imported\n"
+ + "entryUUID: " + IMPORTED_ENTRY_UUID + "\n"
+ + "\n",
+ };
+ }
+
+ private static DN dnOf(String ldif)
+ {
+ return DN.valueOf(ldif.substring("dn: ".length(), ldif.indexOf('\n')));
+ }
+
+ /** The records of the error log which carry the provided message id and the provided CSN. */
+ private static List<String> errorLogRecordsOf(int msgId, CSN csn)
+ {
+ final List<String> records = new ArrayList<>();
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ if (record.contains("msgID=" + msgId) && record.contains(csn.toString()))
+ {
+ records.add(record);
+ }
+ }
+ return records;
+ }
+
+ private void replayMsg(UpdateMsg updateMsg) throws InterruptedException
+ {
+ domain.processUpdate(updateMsg);
+ final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage();
+ domain.markInProgress(ldapUpdate);
+ domain.replay(ldapUpdate, SHUTDOWN);
+ }
+}
--
Gitblit v1.10.0