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() {
