[#889] Keep a change the replay could not apply out of the ServerState

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.
 

opendj-doc-generated-ref/src/main/docbkx/admin-guide/chap-monitoring.xml

@@ -983,6 +983,14 @@ $ dsconfig
      automatically.</para>
     </listitem>
    </varlistentry>
+   <varlistentry>
+    <term><literal>org.opends.server.replication.UnreplayedChange</literal></term>
+    <listitem>
+     <para>Multimaster replication gave up on a change it could not replay.
+     This replica now diverges from the rest of the topology and must be
+     reinitialized.</para>
+    </listitem>
+   </varlistentry>
    <varlistentry>
     <term><literal>org.opends.server.UncaughtException</literal></term>
     <listitem>

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;
 
@@ -140,6 +141,10 @@ public int getDependentChangesSize()
   /**
    * Add a new LDAPUpdateMsg that was received from the replication server
    * to the pendingList.
+   * <p>
+   * 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.
@@ -152,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
     {
@@ -200,12 +205,71 @@ public void commit(CSN csn)
     }
   }
 
-  public void markInProgress(LDAPUpdateMsg msg)
+  /**
+   * Forgets the changes which have not been replayed yet, without updating the
+   * ServerState.
+   * <p>
+   * 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.
+   * <p>
+   * 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 clearUncommitted()
+  {
+    pendingChangesWriteLock.lock();
+    dependentChangesLock.lock();
+    try
+    {
+      final Iterator<PendingChange> it = pendingChanges.values().iterator();
+      while (it.hasNext())
+      {
+        final PendingChange change = it.next();
+        if (!change.isCommitted())
+        {
+          dependentChanges.remove(change);
+          activeAndDependentChanges.remove(change);
+          it.remove();
+        }
+      }
+    }
+    finally
+    {
+      dependentChangesLock.unlock();
+      pendingChangesWriteLock.unlock();
+    }
+  }
+
+  /**
+   * Marks the change of the provided message as being replayed.
+   *
+   * @param msg
+   *          the message whose change is being replayed
+   * @return {@code false} if this message is not the delivery which is listed as
+   *         pending, which happens when the session was restarted after a failed
+   *         replay while this message was still waiting in the replay queue: the
+   *         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
     {

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
           {

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

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
+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. \
+ The change is being skipped: this replica now diverges from the rest of the topology and must \
+ be reinitialized

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<String, AtomicInteger> 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,

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<Control> createShortCircuitControlList(int resultCode, String
   /** Registered short circuits for operations regardless of controls. */
   private static Map<String, Integer> shortCircuits = new ConcurrentHashMap<>();
 
+  /** How many times a registered short circuit was reached. */
+  private static final Map<String, AtomicInteger> shortCircuitCounts = new ConcurrentHashMap<>();
+
+  /** How many times a registered short circuit must be applied, when it is limited. */
+  private static final Map<String, Integer> shortCircuitLimits = new ConcurrentHashMap<>();
+
+  /**
+   * Returns how many times the short circuit registered for the given operation type and
+   * plugin point was reached. A short circuit registered for a limited number of times is
+   * counted as reached by the operations it let through once that number was used up.
+   *
+   * @param operation The type of operation the short circuit applies to.
+   * @param section The plugin point the short circuit applies to.
+   * @return the number of operations which reached the short circuit
+   */
+  public static int getShortCircuitCount(OperationType operation, String section)
+  {
+    final AtomicInteger count = shortCircuitCounts.get(operation + "/" + section.toLowerCase());
+    return count != null ? count.get() : 0;
+  }
+
+  /**
+   * 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,13 +715,32 @@ 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.
    * @param section The plugin point the short circuit applies to.
    */
   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);
   }
 }