From e62617b5a9b2114d0d559ffca501be61ca2bda46 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 20 Aug 2026 16:37:05 +0300 Subject: [PATCH 1/2] [#889] Keep a change the replay could not apply out of the ServerState A change whose replay failed with anything other than NO_OPERATION, BUSY or UNAVAILABLE was recorded as replayed: the ServerState advanced past it, the replication server never sent it again, and an assured SAFE_READ ack went back to the originating master as if the change had been applied. The replica silently diverged while reporting itself fully caught up. The four solveNamingConflict() overloads collapsed two different outcomes into one "return true": the operation became a no-op after conflict resolution, and the operation simply failed. They now report a ConflictResolution, so replay() can tell them apart, and the four copies of the error log move to the single place which decides what to do. A failure of the server itself - the backend being offline or rebuilt, or the storage failing to serve the operation, which BackendImpl reports with the server-error-result-code - is now retried like an unavailable backend and, if it keeps failing, left out of the ServerState: the replication server still owns the change, so the session is restarted and the change is delivered and replayed again. Every failed replay reports the error in the ack, so an assured write is no longer told it is durable here, and counts in the new replayed-updates-failed monitor attribute. A change which can never be applied on this replica would otherwise stop it for good, so after MAX_REPLAY_ATTEMPTS deliveries it is skipped - but loudly, with the new UnreplayedChange alert telling the administrator that this replica has diverged and must be reinitialized. Failures which are not the server's fault keep being skipped as before, with the same ack, counter and alert. Tests: RemotePendingChangesTest covers the bookkeeping the fix relies on, UpdateOperationTest.failedReplayIsNotRecordedAsReplayed covers the redelivery and the give-up, and a new test in AssuredReplicationPluginTest covers the error ack, which the upstream TODO left untested. --- .../plugin/LDAPReplicationDomain.java | 388 +++++++++++++----- .../plugin/RemotePendingChanges.java | 26 ++ .../opends/server/util/ServerConstants.java | 20 + .../opends/messages/replication.properties | 6 + .../replication/UpdateOperationTest.java | 99 +++++ .../plugin/AssuredReplicationPluginTest.java | 70 +++- .../plugin/RemotePendingChangesTest.java | 112 +++++ 7 files changed, 601 insertions(+), 120 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 2653da06fc..de7f7f0fed 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -310,6 +310,21 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { new AtomicInteger(); /** The number of updates replayed successfully by the replication. */ private final AtomicInteger numReplayedPostOpCalled = new AtomicInteger(); + /** + * How many times a change which could not be replayed is asked for again before this + * replica gives up on it and moves on to the changes which follow it. + */ + private static final int MAX_REPLAY_ATTEMPTS = 3; + /** The number of updates which could not be replayed. */ + private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger(); + /** Set while a replay thread is restarting the session after a failed replay. */ + private final AtomicBoolean replayFailureRecovery = new AtomicBoolean(); + /** Guards {@link #lastFailedCSN} and {@link #lastFailedCSNAttempts}. */ + private final Object replayFailureLock = new Object(); + /** The change on which the last replay failure was seen, guarded by {@link #replayFailureLock}. */ + private CSN lastFailedCSN; + /** How many times it failed in a row, guarded by {@link #replayFailureLock}. */ + private int lastFailedCSNAttempts; private final PersistentServerState state; private volatile boolean generationIdSavedStatus; @@ -2291,6 +2306,7 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown) { Operation op = null; // the last operation on which replay was attempted boolean dependency = false; + boolean replayFailed = false; String replayErrorMsg = null; CSN csn = null; try @@ -2304,7 +2320,7 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown) dependency = remotePendingChanges.checkDependencies(op, msg); boolean replayDone = false; int retryCount = 10; - while (!dependency && !replayDone && retryCount-- > 0) + while (!dependency && !replayDone && !replayFailed && retryCount-- > 0) { if (shutdown.get()) { @@ -2343,7 +2359,9 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown) // was a no-op. For example, an add which has already been // replayed, or a modify DN operation on an entry which has been // renamed by a more recent modify DN. + // The change is in the data: push it to the serverState. replayDone = true; + updateError(csn); } else if (result == ResultCode.BUSY) { @@ -2354,64 +2372,88 @@ else if (result == ResultCode.BUSY) Thread.yield(); continue; } - else if (result == ResultCode.UNAVAILABLE) + else if (isServerFailure(result)) { /* * It can happen when a rebuild is performed or the backend is - * offline (OPENDJ-49). Give the server another chance to process - * this operation after some time. + * offline (OPENDJ-49), or when the storage failed to serve the + * operation. Give the server another chance to process this + * operation after some time. */ Thread.sleep(50); continue; } - else if (op instanceof ModifyOperation) - { - ModifyOperation castOp = (ModifyOperation) op; - dependency = remotePendingChanges.checkDependencies(castOp); - ModifyMsg modifyMsg = (ModifyMsg) msg; - replayDone = !dependency && solveNamingConflict(castOp, modifyMsg); - } - else if (op instanceof DeleteOperation) - { - DeleteOperation castOp = (DeleteOperation) op; - dependency = remotePendingChanges.checkDependencies(castOp); - replayDone = !dependency && solveNamingConflict(castOp, msg); - } - else if (op instanceof AddOperation) - { - AddOperation castOp = (AddOperation) op; - AddMsg addMsg = (AddMsg) msg; - dependency = remotePendingChanges.checkDependencies(castOp); - replayDone = !dependency && solveNamingConflict(castOp, addMsg); - } - else if (op instanceof ModifyDNOperation) - { - ModifyDNOperation castOp = (ModifyDNOperation) op; - ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg; - dependency = remotePendingChanges.checkDependencies(modifyDNMsg); - replayDone = !dependency && solveNamingConflict(castOp, modifyDNMsg); - } - else - { - replayDone = true; // unknown type of operation ?! - } - - if (replayDone) - { - // the update became a dummy update and the result - // of the conflict resolution phase is to do nothing. - // however we still need to push this change to the serverState - updateError(csn); - } else { - /* - * Create a new operation reflecting the new state of the UpdateMsg after conflict resolution - * modified it and try replaying it again. Dependencies might have been replayed by now. - * Note: When msg is a DeleteMsg, the DeleteOperation is properly - * created with subtreeDelete request control when needed. - */ - nextOp = msg.createOperation(conn); + ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO; + if (op instanceof ModifyOperation) + { + ModifyOperation castOp = (ModifyOperation) op; + dependency = remotePendingChanges.checkDependencies(castOp); + ModifyMsg modifyMsg = (ModifyMsg) msg; + resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg); + } + else if (op instanceof DeleteOperation) + { + DeleteOperation castOp = (DeleteOperation) op; + dependency = remotePendingChanges.checkDependencies(castOp); + resolution = dependency ? resolution : solveNamingConflict(castOp, msg); + } + else if (op instanceof AddOperation) + { + AddOperation castOp = (AddOperation) op; + AddMsg addMsg = (AddMsg) msg; + dependency = remotePendingChanges.checkDependencies(castOp); + resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg); + } + else if (op instanceof ModifyDNOperation) + { + ModifyDNOperation castOp = (ModifyDNOperation) op; + ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg; + dependency = remotePendingChanges.checkDependencies(modifyDNMsg); + resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg); + } + // else: unknown type of operation ?! there is nothing to replay + + if (!dependency) + { + switch (resolution) + { + case NOTHING_TO_DO: + // the update became a dummy update and the result + // of the conflict resolution phase is to do nothing. + // however we still need to push this change to the serverState + replayDone = true; + updateError(csn); + break; + + case FAILED: + /* + * The operation did not fail on a naming conflict and not on the server + * either: the change can not be applied on this replica. Skip it so that the + * replica keeps replaying the changes which follow, but report the error in + * the ack and tell the administrator that the data now diverge. + */ + final LocalizableMessage errorMsg = ERR_ERROR_REPLAYING_OPERATION.get( + op, csn, result, op.getErrorMessage()); + logger.error(errorMsg); + replayErrorMsg = errorMsg.toString(); + numFailedReplayedUpdates.incrementAndGet(); + replayDone = true; + skipUnreplayableChange(csn, errorMsg); + break; + + default: + /* + * Create a new operation reflecting the new state of the UpdateMsg after conflict resolution + * modified it and try replaying it again. Dependencies might have been replayed by now. + * Note: When msg is a DeleteMsg, the DeleteOperation is properly + * created with subtreeDelete request control when needed. + */ + nextOp = msg.createOperation(conn); + break; + } + } } } else @@ -2420,17 +2462,35 @@ else if (op instanceof ModifyDNOperation) } } - if (!replayDone && !dependency) + if (!replayDone && !replayFailed && !dependency) { - // Continue with the next change but the servers could now become - // inconsistent. - // Let the repair tool know about this. - final LocalizableMessage message = ERR_LOOP_REPLAYING_OPERATION.get( - op, op.getErrorMessage()); - logger.error(message); - numUnresolvedNamingConflicts.incrementAndGet(); - replayErrorMsg = message.toString(); - updateError(csn); + numFailedReplayedUpdates.incrementAndGet(); + if (isServerFailure(op.getResultCode())) + { + /* + * The server kept failing to apply the change, so the change is not in the data. + * Leave it out of the ServerState, otherwise the replication server would never + * send it again and this replica would silently diverge while reporting itself + * up to date. + */ + final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get( + op, csn, op.getResultCode(), op.getErrorMessage()); + logger.error(message); + replayErrorMsg = message.toString(); + replayFailed = true; + } + else + { + // Conflict resolution kept rewriting an operation which kept failing. + // Continue with the next change but the servers could now become inconsistent. + // Let the repair tool know about this. + final LocalizableMessage message = ERR_LOOP_REPLAYING_OPERATION.get( + op, op.getErrorMessage()); + logger.error(message); + numUnresolvedNamingConflicts.incrementAndGet(); + replayErrorMsg = message.toString(); + skipUnreplayableChange(csn, message); + } } } catch (DecodeException | LDAPException | DataFormatException e) { @@ -2440,9 +2500,8 @@ else if (op instanceof ModifyDNOperation) if (csn != null) { /* - * An Exception happened during the replay process. - * Continue with the next change but the servers will now start - * to be inconsistent. + * An Exception happened during the replay process: the change is not in the + * data, so it must not be recorded as replayed. * Let the repair tool know about this. */ LocalizableMessage message = @@ -2450,7 +2509,8 @@ else if (op instanceof ModifyDNOperation) stackTraceToSingleLineString(e), op); logger.error(message); replayErrorMsg = message.toString(); - updateError(csn); + numFailedReplayedUpdates.incrementAndGet(); + replayFailed = true; } else { replayErrorMsg = logDecodingOperationError(msg, e); @@ -2463,6 +2523,13 @@ else if (op instanceof ModifyDNOperation) } } + if (replayFailed && recoverFromReplayFailure(csn, shutdown)) + { + // The ack has been published and the change, still owned by the replication + // server, is being delivered again: there is nothing left to replay here. + return; + } + // Now replay any pending update that had a dependency and whose // dependency has been replayed, do that until no more updates of that // type left... @@ -2505,6 +2572,105 @@ private void updateError(CSN csn) } } + /** + * Returns whether the provided result code reports a failure of this server rather + * than a change which can not be applied: the backend being offline or rebuilt + * (OPENDJ-49), or the storage failing to serve the operation. + * + * @param result the result code of a replayed operation + * @return {@code true} if the operation failed on the server itself + */ + private boolean isServerFailure(ResultCode result) + { + return result == ResultCode.UNAVAILABLE + || result == getServerContext().getCoreConfigManager().getServerErrorResultCode(); + } + + /** + * Records a change which could not be replayed as replayed anyway, so that this replica + * keeps replaying the changes which follow it, and warns that the data now diverge. + * + * @param csn the CSN of the change which could not be replayed + * @param cause the message describing why it could not be replayed + */ + private void skipUnreplayableChange(CSN csn, LocalizableMessage cause) + { + updateError(csn); + DirectoryServer.sendAlertNotification( + this, ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, cause); + } + + /** + * Recovers from a change which could not be replayed. + *

+ * The change has deliberately been left out of the ServerState, so the replication + * server still owns it: restart the session so that it is sent again and replayed on + * a backend which has hopefully recovered in the meantime. Give up after + * {@link #MAX_REPLAY_ATTEMPTS} deliveries of the same change and record it as + * replayed, so that a change which can never be applied here does not stop this + * replica for good: the administrator is told that this replica has diverged and + * must be reinitialized. + * + * @param csn + * the CSN of the change which could not be replayed, {@code null} when the + * message could not even be decoded + * @param shutdown + * whether the server initiated shutdown + * @return {@code true} when the caller must stop replaying because the session is + * being restarted or is going away, {@code false} when it may carry on with + * the changes which follow + */ + private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean shutdown) + { + if (csn == null) + { + // The message could not be decoded: there is nothing to ask for again. + return false; + } + if (shutdown.get() || disabled) + { + // There is no session to have the change sent over again. + return true; + } + + final int attempts; + synchronized (replayFailureLock) + { + lastFailedCSNAttempts = csn.equals(lastFailedCSN) ? lastFailedCSNAttempts + 1 : 1; + lastFailedCSN = csn; + attempts = lastFailedCSNAttempts; + } + + if (attempts > MAX_REPLAY_ATTEMPTS) + { + final LocalizableMessage message = + ERR_REPLAY_SKIPPING_CHANGE.get(csn, getBaseDN(), attempts); + logger.error(message); + skipUnreplayableChange(csn, message); + return false; + } + + logger.error(ERR_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), attempts); + if (!replayFailureRecovery.compareAndSet(false, true)) + { + // Another replay thread is already restarting the session. + return true; + } + try + { + disableService(); + // The uncommitted changes are about to be sent again: forget the ones still listed + // as pending, or processUpdate() would discard them as duplicates. + remotePendingChanges.clear(); + enableService(); + } + finally + { + replayFailureRecovery.set(false); + } + return true; + } + /** * Generate a new CSN and insert it in the pending list. * @@ -2581,14 +2747,25 @@ private DN findEntryDN(String uuid) return null; } + /** Outcome of the conflict resolution attempted after a replayed operation failed. */ + private enum ConflictResolution + { + /** The update message was adjusted: the operation must be replayed again. */ + REPLAY_AGAIN, + /** The change is already reflected in the data: there is nothing left to replay. */ + NOTHING_TO_DO, + /** The operation failed for a reason which is not a naming conflict. */ + FAILED + } + /** * Solve a conflict detected when replaying a modify operation. * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. - * @return true if the process is completed, false if it must continue.. + * @return the outcome of the conflict resolution */ - private boolean solveNamingConflict(ModifyOperation op, ModifyMsg msg) + private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg) { ResultCode result = op.getResultCode(); ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT); @@ -2609,14 +2786,14 @@ private boolean solveNamingConflict(ModifyOperation op, ModifyMsg msg) // replay the modify using the current dn of this entry. msg.setDN(newDN); numResolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } else { // This entry does not exist anymore. // It has probably been deleted, stop the processing of this operation numResolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } } else if (result == ResultCode.NOT_ALLOWED_ON_RDN) @@ -2631,7 +2808,7 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN) { // The entry does not exist anymore. numResolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } // The modify operation is trying to delete the value that is @@ -2657,15 +2834,13 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN) } msg.setMods(mods); numResolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } else { - // The other type of errors can not be caused by naming conflicts. - // Log a message for the repair tool. - logger.error(ERR_ERROR_REPLAYING_OPERATION, - op, ctx.getCSN(), result, op.getErrorMessage()); - return true; + // The other type of errors can not be caused by naming conflicts: + // the operation simply failed, replay() reports it. + return ConflictResolution.FAILED; } } @@ -2674,9 +2849,9 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN) * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. - * @return true if the process is completed, false if it must continue.. + * @return the outcome of the conflict resolution */ - private boolean solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) + private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) { ResultCode result = op.getResultCode(); DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT); @@ -2695,14 +2870,14 @@ private boolean solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) * In any case, there is nothing more to do. */ numResolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } else { // This entry has been renamed, replay the delete using its new DN. msg.setDN(currentDN); numResolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } } else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF) @@ -2722,15 +2897,13 @@ else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF) numUnresolvedNamingConflicts.incrementAndGet(); } - return false; + return ConflictResolution.REPLAY_AGAIN; } else { - // The other type of errors can not be caused by naming conflicts. - // Log a message for the repair tool. - logger.error(ERR_ERROR_REPLAYING_OPERATION, - op, ctx.getCSN(), result, op.getErrorMessage()); - return true; + // The other type of errors can not be caused by naming conflicts: + // the operation simply failed, replay() reports it. + return ConflictResolution.FAILED; } } @@ -2739,10 +2912,10 @@ else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF) * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. - * @return true if the process is completed, false if it must continue. + * @return the outcome of the conflict resolution * @throws Exception When the operation is not valid. */ -private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) +private ConflictResolution solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) throws Exception { ResultCode result = op.getResultCode(); @@ -2788,7 +2961,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) { markConflictEntry(op, currentDN, currentDN.parent().child(newRDN)); numUnresolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } DN newDN = newSuperior.child(newRDN); @@ -2801,7 +2974,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) // The entry has been deleted, we can safely assume // that the operation is completed. numResolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } // if the newDN and the current DN match then the operation @@ -2810,7 +2983,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) if (newDN.equals(currentDN)) { numResolvedNamingConflicts.incrementAndGet(); - return true; + return ConflictResolution.NOTHING_TO_DO; } if (result == ResultCode.NO_SUCH_OBJECT @@ -2825,7 +2998,7 @@ private boolean solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) modifyDnMsg.setDN(currentDN); modifyDnMsg.setNewSuperior(newSuperior.toString()); numResolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } else if (result == ResultCode.ENTRY_ALREADY_EXISTS) { @@ -2841,15 +3014,13 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS) modifyDnMsg.getNewRDN())); modifyDnMsg.setNewSuperior(newSuperior.toString()); numUnresolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } else { - // The other type of errors can not be caused by naming conflicts. - // Log a message for the repair tool. - logger.error(ERR_ERROR_REPLAYING_OPERATION, - op, ctx.getCSN(), result, op.getErrorMessage()); - return true; + // The other type of errors can not be caused by naming conflicts: + // the operation simply failed, replay() reports it. + return ConflictResolution.FAILED; } } @@ -2858,10 +3029,10 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS) * * @param op The operation that triggered the conflict detection. * @param msg The message that triggered the conflict detection. - * @return true if the process is completed, false if it must continue. + * @return the outcome of the conflict resolution * @throws Exception When the operation is not valid. */ - private boolean solveNamingConflict(AddOperation op, AddMsg msg) + private ConflictResolution solveNamingConflict(AddOperation op, AddMsg msg) throws Exception { ResultCode result = op.getResultCode(); @@ -2884,7 +3055,7 @@ private boolean solveNamingConflict(AddOperation op, AddMsg msg) * message for the repair tool to look at this problem. * TODO : Log the message */ - return true; + return ConflictResolution.NOTHING_TO_DO; } DN parentDn = findEntryDN(parentUniqueId); if (parentDn == null) @@ -2911,7 +3082,7 @@ private boolean solveNamingConflict(AddOperation op, AddMsg msg) msg.setDN(DN.valueOf(msg.getDN().rdn() + "," + parentDn)); numResolvedNamingConflicts.incrementAndGet(); } - return false; + return ConflictResolution.REPLAY_AGAIN; } else if (result == ResultCode.ENTRY_ALREADY_EXISTS) { @@ -2927,7 +3098,7 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS) if (findEntryDN(entryUUID) != null) { // entry already exist : this is a replay - return true; + return ConflictResolution.NOTHING_TO_DO; } else { @@ -2936,16 +3107,14 @@ else if (result == ResultCode.ENTRY_ALREADY_EXISTS) generateConflictRDN(entryUUID, msg.getDN().toString()); msg.setDN(DN.valueOf(conflictRDN)); numUnresolvedNamingConflicts.incrementAndGet(); - return false; + return ConflictResolution.REPLAY_AGAIN; } } else { - // The other type of errors can not be caused by naming conflicts. - // log a message for the repair tool. - logger.error(ERR_ERROR_REPLAYING_OPERATION, - op, ctx.getCSN(), result, op.getErrorMessage()); - return true; + // The other type of errors can not be caused by naming conflicts: + // the operation simply failed, replay() reports it. + return ConflictResolution.FAILED; } } @@ -3847,6 +4016,8 @@ public Map getAlerts() alerts.put(ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, ALERT_DESCRIPTION_REPLICATION_UNRESOLVED_CONFLICT); + alerts.put(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, + ALERT_DESCRIPTION_REPLICATION_UNREPLAYED_CHANGE); return alerts; } @@ -4301,6 +4472,7 @@ public void addAdditionalMonitoring(MonitorData attributes) { attributes.add("pending-updates", pendingChanges.size()); attributes.add("replayed-updates-ok", numReplayedPostOpCalled); + attributes.add("replayed-updates-failed", numFailedReplayedUpdates); attributes.add("resolved-modify-conflicts", numResolvedModifyConflicts); attributes.add("resolved-naming-conflicts", numResolvedNamingConflicts); attributes.add("unresolved-naming-conflicts", numUnresolvedNamingConflicts); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java index 06d37970f0..8b925fe752 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java @@ -13,6 +13,7 @@ * * Copyright 2007-2009 Sun Microsystems, Inc. * Portions Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; @@ -200,6 +201,31 @@ public void commit(CSN csn) } } + /** + * Forgets every pending change without updating the ServerState. + *

+ * Called when the replay of a change failed and the session to the + * replication server is restarted: the changes that were not committed must + * be replayed again, and {@link #putRemoteUpdate(LDAPUpdateMsg)} would + * otherwise discard the ones still listed here as duplicates. + */ + public void clear() + { + pendingChangesWriteLock.lock(); + dependentChangesLock.lock(); + try + { + pendingChanges.clear(); + dependentChanges.clear(); + activeAndDependentChanges.clear(); + } + finally + { + dependentChangesLock.unlock(); + pendingChangesWriteLock.unlock(); + } + } + public void markInProgress(LDAPUpdateMsg msg) { pendingChangesReadLock.lock(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java b/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java index ff7f3f13bb..999bef3eaf 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/util/ServerConstants.java @@ -3064,6 +3064,26 @@ public final class ServerConstants "org.opends.server.replication.UnresolvedConflict"; + /** + * The description for the alert type string that will be used for the alert + * notification generated when multimaster replication gives up on a change + * it could not replay. + */ + public static final String ALERT_DESCRIPTION_REPLICATION_UNREPLAYED_CHANGE = + "This alert type will be used to notify administrators when " + + "multimaster replication gives up on a change it could not replay, " + + "leaving this replica diverging from the rest of the topology."; + + + /** + * The alert type string that will be used for the alert notification + * generated when multimaster replication gives up on a change it could not + * replay. + */ + public static final String ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE = + "org.opends.server.replication.UnreplayedChange"; + + /** * The extensible indexer identifier string that will be used for a substring diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index 367a21191f..1237c1c1b7 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -610,3 +610,9 @@ ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port w last attempt to bind it NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \ in domain "%s" from this directory server DS(%d): rejecting the request from the remote directory server DS(%d): %s +ERR_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (attempt %d). \ + The change has not been recorded as replayed: restarting the session to the replication server \ + so that it is sent again +ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s" after %d attempts. \ + The change is being skipped: this replica now diverges from the rest of the topology and must \ + be reinitialized diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index d973d2d136..9e9d48b624 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -50,6 +50,8 @@ import org.opends.server.plugins.ShortCircuitPlugin; import org.opends.server.replication.common.CSN; import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.plugin.LDAPReplicationDomain; +import org.opends.server.replication.plugin.MultimasterReplication; import org.opends.server.replication.protocol.AddMsg; import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.HeartbeatThread; @@ -1371,6 +1373,103 @@ public void call() throws Exception } } + /** + * Test case for [Issue 889]: a change whose replay failed on the server itself must + * not be recorded as replayed. Recording it would advance the ServerState past the + * change, so the replication server would never send it again while this replica + * reports itself up to date. + */ + @Test + public void failedReplayIsNotRecordedAsReplayed() throws Exception + { + testSetUp("failedReplayIsNotRecordedAsReplayed"); + logger.error(LocalizableMessage.raw("Starting replication test : failedReplayIsNotRecordedAsReplayed")); + + final int serverId = 12; + ReplicationBroker broker = + openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry tmp = TestCaseUtils.addEntry( + "dn: uid=user.889," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.889", + "cn: Aaccf Amar", + "sn: Amar"); + String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed"); + final int initialAlerts = DummyAlertHandler.getAlertCount(); + + /* + * Fail the replay the way a storage failure does: the backend reports it with the + * server-error-result-code, 80 by default. The short circuit has to be set at the + * pre-parse plugin point, the pre-operation ones are not invoked for synchronization + * operations. + */ + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); + try + { + final CSN csn = gen.newCSN(); + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + + /* + * The replication server resumes from the ServerState of this replica, so it only + * sends the change again as long as the state does not cover it: seeing the same + * change fail more than once is what tells that it was not recorded as replayed. + */ + TestTimer timer = new TestTimer.Builder() + .maxSleep(60, SECONDS) + .sleepTimes(100, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(getMonitorAttrValue(baseDN, "replayed-updates-failed") >= initialFailures + 2, + "the change was not sent again after its replay failed"); + } + }); + assertNotNull(getEntry(tmp.getName(), 1, true), "the entry must not have been deleted"); + + /* + * The change can never be applied here, so the replica eventually gives up on it + * rather than stopping for good: it then warns that it has diverged. + */ + TestTimer giveUpTimer = new TestTimer.Builder() + .maxSleep(60, SECONDS) + .sleepTimes(200, MILLISECONDS) + .toTimer(); + giveUpTimer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(domain.getServerState().cover(csn), + "the replica did not give up on a change it can never replay"); + } + }); + Assertions.assertThat(DummyAlertHandler.getAlertCount()).isGreaterThan(initialAlerts); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + } + } + finally + { + broker.stop(); + } + } + /** * Enable or disable the receive status of a synchronization provider. * diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java index f59bce4cf4..b5b31b4e5f 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java @@ -50,6 +50,7 @@ import org.opends.server.core.AddOperation; import org.opends.server.core.DeleteOperation; import org.opends.server.core.DirectoryServer; +import org.opends.server.plugins.ShortCircuitPlugin; import org.opends.server.protocols.internal.InternalSearchOperation; import org.opends.server.protocols.internal.SearchRequest; import org.opends.server.replication.ReplicationTestCase; @@ -73,6 +74,7 @@ import org.opends.server.types.Attribute; import org.opends.server.types.Entry; import org.opends.server.types.Operation; +import org.opends.server.types.OperationType; import org.opends.server.types.SearchFilter; import org.opends.server.types.SearchResultEntry; import org.opends.server.util.StaticUtils; @@ -1170,22 +1172,66 @@ public void testSafeReadModeReply(byte rsGroupId) throws Exception return; } + } finally + { + endTest(testcase); + } + } + + /** + * Tests that a DS which could not replay an update in safe read mode acks the replay + * error instead of reporting the update as applied (issue #889). + */ + @Test + public void testSafeReadModeReplyWithReplayError() throws Exception + { + int TIMEOUT = 5000; + String testcase = "testSafeReadModeReplyWithReplayError"; + try + { + // Create and start a RS expecting clients in safe read assured mode + replicationServer = new FakeReplicationServer((byte) 1, replServerPort, RS_SERVER_ID, + true, testcase); + replicationServer.start(NO_READ); - /* Send an update with error from the RS and get the ack with error */ + safeReadDomainCfgEntry = createAssuredDomain(AssuredMode.SAFE_READ_MODE, 0, TIMEOUT); + waitForConnectionToRs(testcase, replicationServer); + + Entry entry = makeEntry( + "dn: ou=assured-sr-replay-error-entry," + SAFE_READ_DN, + "objectClass: top", + "objectClass: organizationalUnit"); + String parentUid = getEntryUUID(DN.valueOf(SAFE_READ_DN)); + + /* + * Fail the replay the way a storage failure does: the backend reports it with the + * server-error-result-code, 80 by default. The short circuit has to be set at the + * pre-parse plugin point, the pre-operation ones are not invoked for synchronization + * operations. + */ + ShortCircuitPlugin.registerShortCircuit( + OperationType.ADD, "PreParse", ResultCode.OTHER.intValue()); + try + { + AckMsg ackMsg = replicationServer.sendAssuredAddMsg(entry, parentUid); - // Make the RS send a not possible assured add message + assertNull(DirectoryServer.getEntry(entry.getName()), "the entry must not have been added"); - // TODO: make the domain return an error: use a plugin ? - // The resolution code does not generate any error so we need to find a - // way to have the replay not working to test this... + // Check that DS replied an ack reporting the replay error + assertFalse(ackMsg.hasTimeout()); + assertTrue(ackMsg.hasReplayError(), "the ack must report the failed replay"); + assertFalse(ackMsg.hasWrongStatus()); + Assertions.assertThat(ackMsg.getFailedServers()).containsExactly(1); - // Check that DS replied an ack with errors -// assertFalse(ackMsg.hasTimeout()); -// assertTrue(ackMsg.hasReplayError()); -// assertFalse(ackMsg.hasWrongStatus()); -// List failedServers = ackMsg.getFailedServers(); -// assertEquals(failedServers.size(), 1); -// assertEquals((integer)failedServers.get(0), (integer)1); + /* + * No monitoring assertion here: the domain restarts its session to have the change + * sent again, which takes its monitor entry away and resets the assured counters. + */ + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.ADD, "PreParse"); + } } finally { endTest(testcase); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java new file mode 100644 index 0000000000..01bdd75ad6 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java @@ -0,0 +1,112 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import static org.testng.Assert.*; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.DeleteMsg; +import org.testng.annotations.Test; + +/** + * Tests the bookkeeping a replica does on the changes it received from a replication + * server: a change reaches the ServerState only once it really has been replayed. + */ +@SuppressWarnings("javadoc") +public class RemotePendingChangesTest extends ReplicationTestCase +{ + private static final int SERVER_ID = 42; + + @Test + public void committedChangeIsPushedToTheServerState() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN(); + + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1"))); + assertEquals(pendingChanges.getQueueSize(), 1); + + pendingChanges.commit(csn); + + assertTrue(state.cover(csn)); + assertEquals(pendingChanges.getQueueSize(), 0); + } + + /** + * A change which is not committed must hold back the ServerState, even when the + * changes which follow it have been replayed: the replication server resumes from the + * ServerState, so anything it covers is never sent again. + */ + @Test + public void uncommittedChangeHoldsBackTheChangesWhichFollowIt() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN failed = generator.newCSN(); + final CSN next = generator.newCSN(); + + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1"))); + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2"))); + + // The replay of the first change failed, the second one went through. + pendingChanges.commit(next); + + assertFalse(state.cover(failed), "a change which was not replayed must not be covered"); + assertFalse(state.cover(next), "the changes which follow a failed one must not be covered either"); + assertEquals(pendingChanges.getQueueSize(), 2); + + // The first change finally made it: both are now recorded as replayed. + pendingChanges.commit(failed); + + assertTrue(state.cover(failed)); + assertTrue(state.cover(next)); + assertEquals(pendingChanges.getQueueSize(), 0); + } + + /** + * Restarting the session forgets the pending changes without recording them, so that + * the ones the replication server sends again are replayed rather than discarded as + * duplicates. + */ + @Test + public void clearForgetsThePendingChangesWithoutRecordingThem() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN(); + + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1"))); + assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "a duplicate must be discarded"); + + pendingChanges.clear(); + + assertEquals(pendingChanges.getQueueSize(), 0); + assertTrue(state.isEmpty(), "clearing the pending changes must not record them as replayed"); + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), + "the change must be accepted again once the session has been restarted"); + } + + private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception + { + return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID); + } +} From 8d3a0cc223c1094477369ee456fa98546babc5e3 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 21 Aug 2026 13:12:10 +0300 Subject: [PATCH 2/2] [#889] Count the replay failures of each change and forget only what was not replayed The count of failed replays lived in two scalar fields, so a backend which fails every change in flight had each of them reset the count of the previous one: the give up after MAX_REPLAY_ATTEMPTS was never reached and the replica restarted its session to the replication server without end. The count is kept per CSN now and dropped as soon as the change is replayed or given up on, and the session is left down for a moment before the change is asked for again. Restarting the session forgot every pending change, including the ones replayed while an older change was failing: those are not in the ServerState yet, so the replication server sends them again and an empty pending list had them replayed and acked a second time - the duplicate check of OPENDJ-1115. Only the changes which were not replayed are forgotten now, and putRemoteUpdate() no longer overwrites the copy which is listed with the one the new delivery came with, which lost the fact that it had been replayed. A message which is not the delivery listed as pending is dropped rather than replayed: it was waiting in the replay queue, shared by every domain, while the session was restarted. markInProgress() reports it instead of adding a null to activeAndDependentChanges, which threw an NPE into the replay thread and left the assured ack unsent. A duplicate delivery no longer has the listener push its CSN to the ServerState either: the copy which is listed owns the change and records it once it really has been replayed, while the ack and the window credit stay per delivery. The guard of the recovery read the AtomicBoolean of the replay thread, which shadowed the field of the domain, and the "disabled" flag which shutdown() does not set: a replay failing while the domain was being shut down brought a broker and a listener thread back up on it. It reads the state of the domain now, and reads it again after the wait. isServerFailure() no longer takes a change away from conflict resolution: server-error-result-code is configurable and is not validated as a result code, so the codes solveNamingConflict() solves are excluded from it. replayed-updates-failed counts changes really skipped, not attempts; the change is skipped on the MAX_REPLAY_ATTEMPTS-th attempt and the message says so; the retry is logged as WARN_REPLAY_RETRYING_CHANGE; the UnreplayedChange alert is documented in the admin guide and is not raised again for a minute, since one cause makes every change in flight unreplayable. Tests: two changes failing at once are both given up on, a failure which clears within the retry window has its change replayed exactly once, committed changes survive a session restart and the previous delivery of a change is not replayed. --- .../asciidoc/admin-guide/chap-monitoring.adoc | 3 + .../docbkx/admin-guide/chap-monitoring.xml | 8 + .../plugin/LDAPReplicationDomain.java | 257 ++++++++++++++---- .../plugin/RemotePendingChanges.java | 54 +++- .../replication/plugin/ReplayThread.java | 11 +- .../opends/messages/replication.properties | 2 +- .../server/extensions/DummyAlertHandler.java | 22 ++ .../server/plugins/ShortCircuitPlugin.java | 69 ++++- .../replication/UpdateOperationTest.java | 185 ++++++++++++- .../plugin/RemotePendingChangesTest.java | 93 ++++++- 10 files changed, 625 insertions(+), 79 deletions(-) diff --git a/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc b/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc index 5adea899e8..ceb1133ac0 100644 --- a/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc +++ b/opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc @@ -2179,6 +2179,9 @@ The directory server detects that its configuration has been manually edited wit `org.opends.server.replication.UnresolvedConflict`:: Multimaster replication cannot resolve a conflict automatically. +`org.opends.server.replication.UnreplayedChange`:: +Multimaster replication gave up on a change it could not replay. This replica now diverges from the rest of the topology and must be reinitialized. + `org.opends.server.UncaughtException`:: A directory server thread has encountered an uncaught exception that caused that thread to terminate abnormally. The impact that this problem has on the server depends on which thread was impacted and the nature of the exception. diff --git a/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml b/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml index e4c85d69bf..077fb12aed 100644 --- a/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml +++ b/opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml @@ -983,6 +983,14 @@ $ dsconfig automatically. + + org.opends.server.replication.UnreplayedChange + + Multimaster replication gave up on a change it could not replay. + This replica now diverges from the rest of the topology and must be + reinitialized. + + org.opends.server.UncaughtException diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index de7f7f0fed..3d666305c4 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -34,6 +34,7 @@ import java.io.OutputStream; import java.io.StringReader; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Date; @@ -51,10 +52,12 @@ import java.util.StringTokenizer; import java.util.TreeMap; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.zip.DataFormatException; @@ -79,6 +82,7 @@ import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy; import org.forgerock.opendj.server.config.server.ExternalChangelogDomainCfg; import org.forgerock.opendj.server.config.server.ReplicationDomainCfg; +import org.forgerock.util.annotations.VisibleForTesting; import org.opends.server.api.AlertGenerator; import org.opends.server.api.DirectoryThread; import org.opends.server.api.LocalBackend; @@ -311,20 +315,53 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { /** The number of updates replayed successfully by the replication. */ private final AtomicInteger numReplayedPostOpCalled = new AtomicInteger(); /** - * How many times a change which could not be replayed is asked for again before this - * replica gives up on it and moves on to the changes which follow it. + * How many times the replay of a change is attempted before this replica gives up on + * it and moves on to the changes which follow it. */ private static final int MAX_REPLAY_ATTEMPTS = 3; - /** The number of updates which could not be replayed. */ + /** + * How long the session is left down before the change is asked for again, multiplied + * by the number of attempts already made: a backend which keeps failing must not be + * hammered with a session restart per failed change. + */ + private static final long REPLAY_RETRY_DELAY_IN_MS = 1000; + /** + * How many changes the replay failures are remembered for. Only the changes which are + * failing are listed here, so the bound is never reached in practice: it is there so + * that a replica failing every change it is sent can not grow the map without end. + */ + private static final int MAX_FAILED_REPLAY_ATTEMPTS_TRACKED = 1000; + /** + * How long the alert telling that this replica diverges is not sent again. A single + * cause - a schema which does not match, a backend which is gone - makes every change + * in flight unreplayable, and one alert per change would be a storm. + */ + private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000; + /** The number of updates this replica gave up replaying. */ private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger(); /** Set while a replay thread is restarting the session after a failed replay. */ private final AtomicBoolean replayFailureRecovery = new AtomicBoolean(); - /** Guards {@link #lastFailedCSN} and {@link #lastFailedCSNAttempts}. */ - private final Object replayFailureLock = new Object(); - /** The change on which the last replay failure was seen, guarded by {@link #replayFailureLock}. */ - private CSN lastFailedCSN; - /** How many times it failed in a row, guarded by {@link #replayFailureLock}. */ - private int lastFailedCSNAttempts; + /** + * How many times the replay of a change failed, keyed on the CSN of the change. + *

+ * The count has to be kept per change: a backend which is failing fails every change + * in flight, and a single slot would be reset by each of them in turn, so the give up + * after {@link #MAX_REPLAY_ATTEMPTS} would never be reached. Entries are dropped as + * soon as the change is replayed or given up on, and the oldest ones are evicted if + * more than {@link #MAX_FAILED_REPLAY_ATTEMPTS_TRACKED} changes are failing at once. + */ + private final ConcurrentSkipListMap failedReplayAttempts = new ConcurrentSkipListMap<>(); + /** When the alert about a change this replica gave up on was last sent. */ + private final AtomicLong lastUnreplayedChangeAlertTime = new AtomicLong(); + /** + * The result codes conflict resolution knows how to solve. The result code the server + * puts on an internal error is configurable and is not validated as a result code, so + * it could be set to one of these: it must never take a change away from + * {@code solveNamingConflict()}, which is the only thing which can solve them. + */ + private static final Set CONFLICT_RESULT_CODES = new HashSet<>(Arrays.asList( + ResultCode.NO_SUCH_OBJECT, ResultCode.ENTRY_ALREADY_EXISTS, + ResultCode.NOT_ALLOWED_ON_RDN, ResultCode.NOT_ALLOWED_ON_NONLEAF)); private final PersistentServerState state; private volatile boolean generationIdSavedStatus; @@ -2016,6 +2053,8 @@ void synchronize(PostOperationOperation op) if (op.isSynchronizationOperation()) { // Replaying a sync operation numReplayedPostOpCalled.incrementAndGet(); + // The change made it, the failures it went through are history. + failedReplayAttempts.remove(curCSN); try { remotePendingChanges.commit(curCSN); @@ -2283,11 +2322,15 @@ public void shutdown() /** * Marks the specified message as the one currently processed by a replay thread. + * * @param msg the message being processed + * @return {@code false} if the change is not pending anymore, which happens when the + * session was restarted after a failed replay while this message was waiting + * in the replay queue: it is sent again, so this copy must not be replayed */ - void markInProgress(LDAPUpdateMsg msg) + boolean markInProgress(LDAPUpdateMsg msg) { - remotePendingChanges.markInProgress(msg); + return remotePendingChanges.markInProgress(msg); } /** @@ -2295,10 +2338,10 @@ void markInProgress(LDAPUpdateMsg msg) * * @param msg * The UpdateMsg to be replayed. - * @param shutdown - * whether the server initiated shutdown + * @param replayThreadShutdown + * whether the replay thread was asked to stop */ - void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown) + void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) { // Try replay the operation, then flush (replaying) any pending operation // whose dependency has been replayed until no more left. @@ -2322,9 +2365,9 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean shutdown) int retryCount = 10; while (!dependency && !replayDone && !replayFailed && retryCount-- > 0) { - if (shutdown.get()) + if (replayThreadShutdown.get() || shutdown.get()) { - // shutdown initiated, let's leave + // Either this replay thread or this domain is going away, let's leave. return; } // Try replay the operation @@ -2438,7 +2481,6 @@ else if (op instanceof ModifyDNOperation) op, csn, result, op.getErrorMessage()); logger.error(errorMsg); replayErrorMsg = errorMsg.toString(); - numFailedReplayedUpdates.incrementAndGet(); replayDone = true; skipUnreplayableChange(csn, errorMsg); break; @@ -2464,7 +2506,6 @@ else if (op instanceof ModifyDNOperation) if (!replayDone && !replayFailed && !dependency) { - numFailedReplayedUpdates.incrementAndGet(); if (isServerFailure(op.getResultCode())) { /* @@ -2509,7 +2550,6 @@ else if (op instanceof ModifyDNOperation) stackTraceToSingleLineString(e), op); logger.error(message); replayErrorMsg = message.toString(); - numFailedReplayedUpdates.incrementAndGet(); replayFailed = true; } else { @@ -2523,7 +2563,7 @@ else if (op instanceof ModifyDNOperation) } } - if (replayFailed && recoverFromReplayFailure(csn, shutdown)) + if (replayFailed && recoverFromReplayFailure(csn, replayThreadShutdown)) { // The ack has been published and the change, still owned by the replication // server, is being delivered again: there is nothing left to replay here. @@ -2552,12 +2592,16 @@ private String logDecodingOperationError(LDAPUpdateMsg msg, Exception e) * called when error or Exceptions happen during the operation replay. * * @param csn the CSN of the operation with error. + * @return {@code false} if the change was not listed in the pending changes anymore, + * so that it has not been recorded as replayed: the replication server sends + * it again. */ - private void updateError(CSN csn) + private boolean updateError(CSN csn) { try { remotePendingChanges.commit(csn); + return true; } catch (NoSuchElementException e) { @@ -2569,6 +2613,7 @@ private void updateError(CSN csn) "LDAPReplicationDomain.updateError: Unable to find remote " + "pending change for CSN %s", csn); } + return false; } } @@ -2582,8 +2627,12 @@ private void updateError(CSN csn) */ private boolean isServerFailure(ResultCode result) { - return result == ResultCode.UNAVAILABLE - || result == getServerContext().getCoreConfigManager().getServerErrorResultCode(); + if (ResultCode.UNAVAILABLE.equals(result)) + { + return true; + } + return result.equals(getServerContext().getCoreConfigManager().getServerErrorResultCode()) + && !CONFLICT_RESULT_CODES.contains(result); } /** @@ -2595,9 +2644,69 @@ private boolean isServerFailure(ResultCode result) */ private void skipUnreplayableChange(CSN csn, LocalizableMessage cause) { - updateError(csn); - DirectoryServer.sendAlertNotification( - this, ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, cause); + failedReplayAttempts.remove(csn); + if (updateError(csn)) + { + numFailedReplayedUpdates.incrementAndGet(); + sendUnreplayedChangeAlert(cause); + } + // Otherwise the change is not listed as pending anymore - the session was restarted + // while it was being replayed - so it has not been skipped: the replication server + // sends it again and this replica gives up on it then. + } + + /** + * Tells the administrator that this replica gave up on a change and now diverges from + * the rest of the topology. + *

+ * Whatever makes a change unreplayable - a schema which does not match, a backend + * which is gone - makes every change in flight unreplayable too, so the alert is not + * sent again for {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS}: each skipped change + * is logged, the alert is there to have the administrator look at the log. + * + * @param cause the message describing why the change could not be replayed + */ + private void sendUnreplayedChangeAlert(LocalizableMessage cause) + { + final long now = System.currentTimeMillis(); + final long lastSent = lastUnreplayedChangeAlertTime.get(); + if (now - lastSent >= UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS + && lastUnreplayedChangeAlertTime.compareAndSet(lastSent, now)) + { + DirectoryServer.sendAlertNotification( + this, ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, cause); + } + } + + /** + * Lets the next change this replica gives up on raise its alert straight away. + *

+ * Only there for the tests which check the alert: they must not be at the mercy of the + * alert another test raised less than + * {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS} ago. + */ + @VisibleForTesting + public void resetUnreplayedChangeAlertThrottle() + { + lastUnreplayedChangeAlertTime.set(0); + } + + /** + * Counts one more failed replay of the change with the provided CSN. + * + * @param csn the CSN of the change whose replay failed + * @return how many times the replay of that change failed in a row + */ + private int countReplayFailure(CSN csn) + { + final int attempts = failedReplayAttempts.merge(csn, 1, Integer::sum); + while (failedReplayAttempts.size() > MAX_FAILED_REPLAY_ATTEMPTS_TRACKED) + { + // Every change in flight is failing: keep the counts of the most recent ones, + // the oldest ones have been given up on long ago. + failedReplayAttempts.pollFirstEntry(); + } + return attempts; } /** @@ -2605,43 +2714,29 @@ private void skipUnreplayableChange(CSN csn, LocalizableMessage cause) *

* The change has deliberately been left out of the ServerState, so the replication * server still owns it: restart the session so that it is sent again and replayed on - * a backend which has hopefully recovered in the meantime. Give up after - * {@link #MAX_REPLAY_ATTEMPTS} deliveries of the same change and record it as - * replayed, so that a change which can never be applied here does not stop this - * replica for good: the administrator is told that this replica has diverged and - * must be reinitialized. + * a backend which has hopefully recovered in the meantime. Give up once its replay + * failed {@link #MAX_REPLAY_ATTEMPTS} times and record it as replayed, so that a + * change which can never be applied here does not stop this replica for good: the + * administrator is told that this replica has diverged and must be reinitialized. * * @param csn * the CSN of the change which could not be replayed, {@code null} when the * message could not even be decoded - * @param shutdown - * whether the server initiated shutdown + * @param replayThreadShutdown + * whether the replay thread was asked to stop * @return {@code true} when the caller must stop replaying because the session is * being restarted or is going away, {@code false} when it may carry on with * the changes which follow */ - private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean shutdown) + private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean replayThreadShutdown) { if (csn == null) { // The message could not be decoded: there is nothing to ask for again. return false; } - if (shutdown.get() || disabled) - { - // There is no session to have the change sent over again. - return true; - } - - final int attempts; - synchronized (replayFailureLock) - { - lastFailedCSNAttempts = csn.equals(lastFailedCSN) ? lastFailedCSNAttempts + 1 : 1; - lastFailedCSN = csn; - attempts = lastFailedCSNAttempts; - } - - if (attempts > MAX_REPLAY_ATTEMPTS) + final int attempts = countReplayFailure(csn); + if (attempts >= MAX_REPLAY_ATTEMPTS) { final LocalizableMessage message = ERR_REPLAY_SKIPPING_CHANGE.get(csn, getBaseDN(), attempts); @@ -2650,18 +2745,44 @@ private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean shutdown) return false; } - logger.error(ERR_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), attempts); + if (replayThreadShutdown.get() || shutdown.get() || disabled) + { + /* + * This replay thread or this whole domain is going away: there is no session to + * have the change sent over again. Restarting the one which is being stopped would + * leave a broker and a listener thread behind on a domain whose alert generator, + * flush thread and RSUpdater are already gone. + */ + return true; + } + + logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), attempts); if (!replayFailureRecovery.compareAndSet(false, true)) { - // Another replay thread is already restarting the session. + // Another replay thread is already restarting the session: the change it could not + // replay is left out of the ServerState too, so it comes back over the new session. return true; } try { disableService(); - // The uncommitted changes are about to be sent again: forget the ones still listed - // as pending, or processUpdate() would discard them as duplicates. - remotePendingChanges.clear(); + /* + * The uncommitted changes are about to be sent again: forget the ones still listed + * as pending, or processUpdate() would discard them as duplicates. The messages + * they came with may still be waiting in the replay queue, which is shared by every + * domain and which no lock lets this thread drain in reasonable time: they are + * dropped when a replay thread takes them out, because markInProgress() only + * accepts the delivery which is listed as pending. + */ + remotePendingChanges.clearUncommitted(); + // Leave the backend some time to recover rather than ask for the change straight + // away: a session restart is not free for the replication server either. + waitBeforeSessionRestart(attempts); + if (shutdown.get() || disabled) + { + // The domain went away while this thread was waiting. + return true; + } enableService(); } finally @@ -2671,6 +2792,25 @@ private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean shutdown) return true; } + /** + * Waits for a while before the session to the replication server is started again, so + * that a backend which keeps failing is not asked for every change it can not apply as + * fast as the replication server can send them. + * + * @param attempts how many times the replay of the change failed already + */ + private void waitBeforeSessionRestart(int attempts) + { + try + { + Thread.sleep(REPLAY_RETRY_DELAY_IN_MS * attempts); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + /** * Generate a new CSN and insert it in the pending list. * @@ -4429,6 +4569,12 @@ public boolean processUpdate(UpdateMsg updateMsg) * are uncommitted changes in the queue and session failover occurs * causing a recovery of all changes since the current committed server * state. See OPENDJ-1115. + * + * The copy which is already listed owns the change: it is the one which records + * it in the ServerState once it really has been replayed. Report this delivery + * as done - the window and the ack are per delivery - but as handled + * asynchronously, so that the listener does not push the CSN to the ServerState + * over a change which is still being replayed or is failing (issue #889). */ if (logger.isTraceEnabled()) { @@ -4436,7 +4582,8 @@ public boolean processUpdate(UpdateMsg updateMsg) "LDAPReplicationDomain.processUpdate: ignoring " + "duplicate change %s", msg.getCSN()); } - return true; + processUpdateDone(msg, null); + return false; } // Put update message into the replay queue diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java index 8b925fe752..7d05837825 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java @@ -141,6 +141,10 @@ public int getDependentChangesSize() /** * Add a new LDAPUpdateMsg that was received from the replication server * to the pendingList. + *

+ * A change which is already listed is left alone: the copy which is here knows + * whether it has been replayed and what it depends on, while the one which comes + * with the message the replication server sent again knows neither. * * @param update The LDAPUpdateMsg that was received from the replication * server and that will be added to the pending list. @@ -153,7 +157,7 @@ public boolean putRemoteUpdate(LDAPUpdateMsg update) try { CSN csn = update.getCSN(); - return pendingChanges.put(csn, new PendingChange(csn, null, update)) == null; + return pendingChanges.putIfAbsent(csn, new PendingChange(csn, null, update)) == null; } finally { @@ -202,22 +206,39 @@ public void commit(CSN csn) } /** - * Forgets every pending change without updating the ServerState. + * Forgets the changes which have not been replayed yet, without updating the + * ServerState. *

* Called when the replay of a change failed and the session to the * replication server is restarted: the changes that were not committed must * be replayed again, and {@link #putRemoteUpdate(LDAPUpdateMsg)} would * otherwise discard the ones still listed here as duplicates. + *

+ * The changes which have already been committed stay: they are held back by + * the change which failed, so the ServerState does not cover them yet and the + * replication server sends them again. Forgetting them would have them + * replayed a second time, which is exactly what the duplicate check of + * {@link #putRemoteUpdate(LDAPUpdateMsg)} is there to prevent (OPENDJ-1115). + * Keeping them is enough: the ServerState is updated over them as soon as the + * change which holds them back commits. */ - public void clear() + public void clearUncommitted() { pendingChangesWriteLock.lock(); dependentChangesLock.lock(); try { - pendingChanges.clear(); - dependentChanges.clear(); - activeAndDependentChanges.clear(); + final Iterator it = pendingChanges.values().iterator(); + while (it.hasNext()) + { + final PendingChange change = it.next(); + if (!change.isCommitted()) + { + dependentChanges.remove(change); + activeAndDependentChanges.remove(change); + it.remove(); + } + } } finally { @@ -226,12 +247,29 @@ public void clear() } } - public void markInProgress(LDAPUpdateMsg msg) + /** + * Marks the change of the provided message as being replayed. + * + * @param msg + * the message whose change is being replayed + * @return {@code false} if this message is not the delivery which is listed as + * pending, which happens when the session was restarted after a failed + * replay while this message was still waiting in the replay queue: the + * change was either forgotten or the replication server sent it again, + * so this copy must not be replayed. + */ + public boolean markInProgress(LDAPUpdateMsg msg) { pendingChangesReadLock.lock(); try { - activeAndDependentChanges.add(pendingChanges.get(msg.getCSN())); + final PendingChange change = pendingChanges.get(msg.getCSN()); + if (change == null || change.getLDAPUpdateMsg() != msg) + { + return false; + } + activeAndDependentChanges.add(change); + return true; } finally { diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java index 44ea968458..4a31d0c1d4 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java @@ -13,6 +13,7 @@ * * Copyright 2006-2008 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; @@ -98,7 +99,15 @@ public void run() // Find replication domain for that update message and mark it as "in progress" updateMsg = updateToreplay.getUpdateMessage(); domain = updateToreplay.getReplicationDomain(); - domain.markInProgress(updateMsg); + if (!domain.markInProgress(updateMsg)) + { + /* + * The domain restarted its session after a failed replay while this + * message was waiting here, so it does not know about this change + * anymore: the replication server sends it again over the new session. + */ + continue; + } } finally { diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index 1237c1c1b7..fbd10a26a5 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -610,7 +610,7 @@ ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port w last attempt to bind it NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \ in domain "%s" from this directory server DS(%d): rejecting the request from the remote directory server DS(%d): %s -ERR_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (attempt %d). \ +WARN_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (attempt %d). \ The change has not been recorded as replayed: restarting the session to the replication server \ so that it is sent again ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s" after %d attempts. \ diff --git a/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java b/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java index 181b6918cb..f395262923 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/extensions/DummyAlertHandler.java @@ -13,10 +13,13 @@ * * Copyright 2008 Sun Microsystems, Inc. * Portions Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.extensions; import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.i18n.LocalizableMessage; @@ -42,6 +45,9 @@ public class DummyAlertHandler /** The number of times this alert handler has been invoked. */ private static AtomicInteger alertCount = new AtomicInteger(0); + /** The number of times this alert handler has been invoked, per alert type. */ + private static final ConcurrentMap alertCountByType = new ConcurrentHashMap<>(); + /** Creates a new instance of this SMTP alert handler. */ public DummyAlertHandler() { @@ -86,6 +92,7 @@ public void sendAlertNotification(AlertGenerator generator, String alertType, LocalizableMessage alertMessage) { alertCount.incrementAndGet(); + alertCountByType.computeIfAbsent(alertType, type -> new AtomicInteger()).incrementAndGet(); } /** @@ -98,6 +105,21 @@ public static int getAlertCount() return alertCount.get(); } + /** + * Retrieves the number of times that this alert handler has been invoked with the + * provided alert type. + * + * @param alertType The type of the alert notifications to count. + * + * @return The number of times that this alert handler has been given an alert + * notification of that type. + */ + public static int getAlertCount(String alertType) + { + final AtomicInteger count = alertCountByType.get(alertType); + return count != null ? count.get() : 0; + } + /** {@inheritDoc} */ @Override public boolean isConfigurationChangeAcceptable(AlertHandlerCfg configuration, diff --git a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java index 2a04473f0a..5dc3cc1eda 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java @@ -13,6 +13,7 @@ * * Copyright 2006-2008 Sun Microsystems, Inc. * Portions Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.plugins; @@ -25,6 +26,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.opendj.config.server.ConfigException; @@ -614,11 +616,18 @@ private int shortCircuitInternal(PluginOperation operation, String section) } // Check for registered short circuits. - Integer resultCode = shortCircuits.get( - operation.getOperationType() + "/" + section.toLowerCase()); + final String key = operation.getOperationType() + "/" + section.toLowerCase(); + Integer resultCode = shortCircuits.get(key); if (resultCode != null) { - return resultCode; + final int reached = shortCircuitCounts.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet(); + final Integer maxTimes = shortCircuitLimits.get(key); + if (maxTimes == null || reached <= maxTimes) + { + return resultCode; + } + // The short circuit was applied as many times as it was asked for: from now on the + // operations are let through, which is how a transient failure is simulated. } // If we've gotten here, then we shouldn't short-circuit the operation @@ -662,6 +671,39 @@ public static List createShortCircuitControlList(int resultCode, String /** Registered short circuits for operations regardless of controls. */ private static Map shortCircuits = new ConcurrentHashMap<>(); + /** How many times a registered short circuit was reached. */ + private static final Map shortCircuitCounts = new ConcurrentHashMap<>(); + + /** How many times a registered short circuit must be applied, when it is limited. */ + private static final Map shortCircuitLimits = new ConcurrentHashMap<>(); + + /** + * Returns how many times the short circuit registered for the given operation type and + * plugin point was reached. A short circuit registered for a limited number of times is + * counted as reached by the operations it let through once that number was used up. + * + * @param operation The type of operation the short circuit applies to. + * @param section The plugin point the short circuit applies to. + * @return the number of operations which reached the short circuit + */ + public static int getShortCircuitCount(OperationType operation, String section) + { + final AtomicInteger count = shortCircuitCounts.get(operation + "/" + section.toLowerCase()); + return count != null ? count.get() : 0; + } + + /** + * Forgets how many times the short circuit registered for the given operation type and + * plugin point was applied. + * + * @param operation The type of operation the short circuit applies to. + * @param section The plugin point the short circuit applies to. + */ + public static void resetShortCircuitCount(OperationType operation, String section) + { + shortCircuitCounts.remove(operation + "/" + section.toLowerCase()); + } + /** * Register a short circuit for the given operation type and plugin point. * @param operation The type of operation the short circuit applies to. @@ -673,6 +715,23 @@ public static void registerShortCircuit(OperationType operation, String section, shortCircuits.put(operation + "/" + section.toLowerCase(), resultCode); } + /** + * Register a short circuit which only applies to the given number of operations, the + * ones which follow being let through: this is how a transient failure is simulated. + * + * @param operation The type of operation the short circuit applies to. + * @param section The plugin point the short circuit applies to. + * @param resultCode The result code to be returned for the short circuit. + * @param maxTimes How many operations must be short circuited. + */ + public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes) + { + final String key = operation + "/" + section.toLowerCase(); + shortCircuitCounts.remove(key); + shortCircuitLimits.put(key, maxTimes); + shortCircuits.put(key, resultCode); + } + /** * Deregister a short circuit for the given operation type and plugin point. * @param operation The type of operation the short circuit applies to. @@ -680,6 +739,8 @@ public static void registerShortCircuit(OperationType operation, String section, */ public static void deregisterShortCircuit(OperationType operation, String section) { - shortCircuits.remove(operation + "/" + section.toLowerCase()); + final String key = operation + "/" + section.toLowerCase(); + shortCircuits.remove(key); + shortCircuitLimits.remove(key); } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index 9e9d48b624..ca1ad016f1 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -26,6 +26,7 @@ import static org.opends.server.protocols.internal.InternalClientConnection.*; import static org.opends.server.replication.plugin.LDAPReplicationDomain.*; import static org.opends.server.util.CollectionUtils.*; +import static org.opends.server.util.ServerConstants.*; import static org.testng.Assert.*; import java.net.SocketTimeoutException; @@ -1405,7 +1406,8 @@ public void failedReplayIsNotRecordedAsReplayed() throws Exception final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed"); - final int initialAlerts = DummyAlertHandler.getAlertCount(); + domain.resetUnreplayedChangeAlertThrottle(); + final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE); /* * Fail the replay the way a storage failure does: the backend reports it with the @@ -1413,6 +1415,7 @@ public void failedReplayIsNotRecordedAsReplayed() throws Exception * pre-parse plugin point, the pre-operation ones are not invoked for synchronization * operations. */ + ShortCircuitPlugin.resetShortCircuitCount(OperationType.DELETE, "PreParse"); ShortCircuitPlugin.registerShortCircuit( OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); try @@ -1423,7 +1426,8 @@ public void failedReplayIsNotRecordedAsReplayed() throws Exception /* * The replication server resumes from the ServerState of this replica, so it only * sends the change again as long as the state does not cover it: seeing the same - * change fail more than once is what tells that it was not recorded as replayed. + * change replayed more than once is what tells that it was not recorded as + * replayed. */ TestTimer timer = new TestTimer.Builder() .maxSleep(60, SECONDS) @@ -1434,7 +1438,7 @@ public void failedReplayIsNotRecordedAsReplayed() throws Exception @Override public void call() throws Exception { - assertTrue(getMonitorAttrValue(baseDN, "replayed-updates-failed") >= initialFailures + 2, + assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse") >= 2, "the change was not sent again after its replay failed"); } }); @@ -1445,7 +1449,7 @@ public void call() throws Exception * rather than stopping for good: it then warns that it has diverged. */ TestTimer giveUpTimer = new TestTimer.Builder() - .maxSleep(60, SECONDS) + .maxSleep(120, SECONDS) .sleepTimes(200, MILLISECONDS) .toTimer(); giveUpTimer.repeatUntilSuccess(new CallableVoid() @@ -1457,7 +1461,178 @@ public void call() throws Exception "the replica did not give up on a change it can never replay"); } }); - Assertions.assertThat(DummyAlertHandler.getAlertCount()).isGreaterThan(initialAlerts); + assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures + 1, + "a change which could not be replayed must be counted once, not once per attempt"); + Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE)) + .as("the administrator must be told that this replica now diverges") + .isGreaterThan(initialAlerts); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + } + } + finally + { + broker.stop(); + } + } + + /** + * Test case for [Issue 889]: every change which can not be replayed must be given up + * on, not only the one which fails on its own. + *

+ * A backend which is failing fails every change in flight, which is what this test + * reproduces with two changes. A count kept for the last failed change only is reset + * by each of them in turn, so the give up would never be reached and this replica + * would restart its session to the replication server without end. + */ + @Test + public void everyChangeWhichCanNotBeReplayedIsGivenUpOn() throws Exception + { + testSetUp("everyChangeWhichCanNotBeReplayedIsGivenUpOn"); + logger.error(LocalizableMessage.raw("Starting replication test : everyChangeWhichCanNotBeReplayedIsGivenUpOn")); + + final int serverId = 13; + ReplicationBroker broker = + openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry first = TestCaseUtils.addEntry( + "dn: uid=user.889.1," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.889.1", + "cn: Aaccf Amar", + "sn: Amar"); + Entry second = TestCaseUtils.addEntry( + "dn: uid=user.889.2," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.889.2", + "cn: Aaccf Amar", + "sn: Amar"); + String firstUuid = getEntry(first.getName(), 1, true).parseAttribute("entryuuid").asString(); + String secondUuid = getEntry(second.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed"); + + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); + try + { + final CSN firstCSN = gen.newCSN(); + final CSN secondCSN = gen.newCSN(); + broker.publish(new DeleteMsg(first.getName(), firstCSN, firstUuid)); + broker.publish(new DeleteMsg(second.getName(), secondCSN, secondUuid)); + + TestTimer giveUpTimer = new TestTimer.Builder() + .maxSleep(120, SECONDS) + .sleepTimes(200, MILLISECONDS) + .toTimer(); + giveUpTimer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(domain.getServerState().cover(firstCSN), + "the replica did not give up on the first change it can never replay"); + assertTrue(domain.getServerState().cover(secondCSN), + "the replica did not give up on the second change it can never replay"); + } + }); + assertNotNull(getEntry(first.getName(), 1, true), "the first entry must not have been deleted"); + assertNotNull(getEntry(second.getName(), 1, true), "the second entry must not have been deleted"); + assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures + 2, + "both changes must be counted as failed, once each"); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + } + } + finally + { + broker.stop(); + } + } + + /** + * Test case for [Issue 889]: a replay which fails on the server itself is retried in + * place, and a failure which clears while it is retried has the change applied exactly + * once, without the session being restarted and without the change being reported as + * failed. + */ + @Test + public void transientReplayFailureIsRetriedAndTheChangeApplied() throws Exception + { + testSetUp("transientReplayFailureIsRetriedAndTheChangeApplied"); + logger.error(LocalizableMessage.raw( + "Starting replication test : transientReplayFailureIsRetriedAndTheChangeApplied")); + + final int serverId = 14; + ReplicationBroker broker = + openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry tmp = TestCaseUtils.addEntry( + "dn: uid=user.889.3," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.889.3", + "cn: Aaccf Amar", + "sn: Amar"); + String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed"); + final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE); + + /* + * The backend is unavailable for the first attempts, the way it is while a rebuild + * is performed or while it is offline (OPENDJ-49), then it serves the operation. + */ + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.UNAVAILABLE.intValue(), 3); + try + { + final CSN csn = gen.newCSN(); + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + + assertNull(getEntry(tmp.getName(), 30000, false), + "the change was not replayed once the backend served the operation again"); + Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")) + .as("the replay must have been retried in place") + .isGreaterThanOrEqualTo(4); + + TestTimer timer = new TestTimer.Builder() + .maxSleep(30, SECONDS) + .sleepTimes(100, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(domain.getServerState().cover(csn), + "a change which was replayed must be recorded as replayed"); + } + }); + assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures, + "a change which was replayed after a transient failure must not count as failed"); + assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts, + "a transient failure must not tell the administrator that this replica diverged"); } finally { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java index 01bdd75ad6..f2b0ead2be 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java @@ -83,12 +83,12 @@ public void uncommittedChangeHoldsBackTheChangesWhichFollowIt() throws Exception } /** - * Restarting the session forgets the pending changes without recording them, so that - * the ones the replication server sends again are replayed rather than discarded as - * duplicates. + * Restarting the session forgets the changes which were not replayed without recording + * them, so that the ones the replication server sends again are replayed rather than + * discarded as duplicates. */ @Test - public void clearForgetsThePendingChangesWithoutRecordingThem() throws Exception + public void clearUncommittedForgetsTheChangesWhichWereNotReplayed() throws Exception { final ServerState state = new ServerState(); final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); @@ -97,7 +97,7 @@ public void clearForgetsThePendingChangesWithoutRecordingThem() throws Exception assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1"))); assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "a duplicate must be discarded"); - pendingChanges.clear(); + pendingChanges.clearUncommitted(); assertEquals(pendingChanges.getQueueSize(), 0); assertTrue(state.isEmpty(), "clearing the pending changes must not record them as replayed"); @@ -105,6 +105,89 @@ public void clearForgetsThePendingChangesWithoutRecordingThem() throws Exception "the change must be accepted again once the session has been restarted"); } + /** + * A change which was replayed while an older one was still failing must survive the + * restart of the session. The ServerState does not cover it yet, so the replication + * server sends it again, and replaying it a second time is exactly what the duplicate + * check of {@code putRemoteUpdate()} is there to prevent (OPENDJ-1115). + */ + @Test + public void clearUncommittedKeepsTheChangesWhichWereReplayed() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN failed = generator.newCSN(); + final CSN next = generator.newCSN(); + + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1"))); + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2"))); + // The replay of the first change failed, the second one went through and is held + // back by the first one. + pendingChanges.commit(next); + + pendingChanges.clearUncommitted(); + + assertEquals(pendingChanges.getQueueSize(), 1); + assertTrue(state.isEmpty(), "clearing the pending changes must not record them as replayed"); + assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(next, "uuid-2")), + "a change which was replayed must not be replayed a second time"); + assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(failed, "uuid-1")), + "the change which was not replayed must be accepted again"); + + // The change which failed finally made it: both are now recorded as replayed. + pendingChanges.commit(failed); + + assertTrue(state.cover(failed)); + assertTrue(state.cover(next)); + assertEquals(pendingChanges.getQueueSize(), 0); + } + + /** + * A message which was waiting in the replay queue while the session was restarted must + * be reported as not pending anymore rather than replayed against a bookkeeping which + * does not list its change. + */ + @Test + public void markInProgressReportsAChangeWhichIsNotPendingAnymore() throws Exception + { + final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState()); + final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN(); + final DeleteMsg msg = deleteMsg(csn, "uuid-1"); + + assertTrue(pendingChanges.putRemoteUpdate(msg)); + assertTrue(pendingChanges.markInProgress(msg)); + assertEquals(pendingChanges.changesInProgressSize(), 1); + + pendingChanges.clearUncommitted(); + + assertEquals(pendingChanges.changesInProgressSize(), 0); + assertFalse(pendingChanges.markInProgress(msg), + "a message whose change was forgotten must not be replayed"); + } + + /** + * The replication server sends the change again over the new session while the message + * of the previous delivery may still be waiting in the replay queue: only the delivery + * which is listed as pending is replayed, or the same change would be applied twice. + */ + @Test + public void markInProgressRejectsThePreviousDeliveryOfAChange() throws Exception + { + final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState()); + final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN(); + final DeleteMsg previousDelivery = deleteMsg(csn, "uuid-1"); + final DeleteMsg newDelivery = deleteMsg(csn, "uuid-1"); + + assertTrue(pendingChanges.putRemoteUpdate(previousDelivery)); + pendingChanges.clearUncommitted(); + assertTrue(pendingChanges.putRemoteUpdate(newDelivery), "the change must be accepted again"); + + assertFalse(pendingChanges.markInProgress(previousDelivery), + "the message of the previous delivery must not be replayed"); + assertTrue(pendingChanges.markInProgress(newDelivery)); + } + private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception { return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID);