mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
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() { 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 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); 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 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
New file @@ -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); } }
