[#818] Drop the empty domain map a replica DB creation leaves behind when it bails out

opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java

@@ -81,6 +81,12 @@ public class FileChangelogDB implements ChangelogDB, ReplicationDomainDB
    * </ol>
    * When creating a replicaDB, synchronize on the domainMap to avoid
    * concurrent shutdown.
+   * <p>
+   * A creation which bails out while holding the domainMap monitor removes the still empty
+   * domainMap it inserted, so that no phantom domain is left behind. It may only do so after
+   * checking, under that same monitor, that the domainMap is still the one mapped to its baseDN:
+   * the removal is equality based and two empty maps are equal, so without the identity check it
+   * could unmap the fresh domainMap of another creation.
    */
   private final ConcurrentMap<DN, ConcurrentMap<Integer, FileReplicaDB>> domainToReplicaDBs =
       new ConcurrentHashMap<>();
@@ -278,25 +284,49 @@ private Pair<FileReplicaDB, Boolean> getExistingOrNewReplicaDB(final ConcurrentM
         // 1) a shutdown was initiated or 2) an initialize was called.
         // Return will allow the code to:
         // 1) shutdown properly or 2) lazily recreate the replicaDB
+        // There is nothing to clean up here: this domainMap is already unmapped, and whatever map
+        // is now associated to baseDN belongs to another creation.
         return null;
       }
 
-      if (shutdown.get())
+      try
       {
-        // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB():
-        // it may already have drained domainToReplicaDBs before this domainMap was inserted into
-        // it, in which case nothing would ever shutdown a replicaDB created here.
-        // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not
-        // created its iterator yet either: since ConcurrentHashMap iterators traverse the
-        // elements as they existed upon construction of the iterator, it will see this domainMap,
-        // which was inserted before this monitor was acquired, and will have to block on this
-        // same monitor to drain it.
-        return null;
-      }
+        if (shutdown.get())
+        {
+          // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB():
+          // it may already have drained domainToReplicaDBs before this domainMap was inserted into
+          // it, in which case nothing would ever shutdown a replicaDB created here.
+          // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not
+          // created its iterator yet either: since ConcurrentHashMap iterators traverse the
+          // elements as they existed upon construction of the iterator, it will see this domainMap,
+          // which was inserted before this monitor was acquired, and will have to block on this
+          // same monitor to drain it.
+          return null;
+        }
 
-      final FileReplicaDB newDB = newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv);
-      domainMap.put(serverId, newDB);
-      return Pair.of(newDB, true);
+        final FileReplicaDB newDB = newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv);
+        domainMap.put(serverId, newDB);
+        return Pair.of(newDB, true);
+      }
+      finally
+      {
+        // Leaving without having created the replica DB must not leave behind the empty domainMap
+        // inserted by getExistingOrNewDomainMap(): nothing would ever remove it, and every multi
+        // domain cursor created afterwards would walk a domain holding no replica DB at all.
+        // Only an empty map may be dropped: a populated one must stay mapped for the drain of
+        // shutdownDB() to find, even when the creation of this serverId failed.
+        // The identity check above passed under this monitor, and every removal site takes the
+        // monitor of the map it unmaps before removing it: the mapping cannot have changed since,
+        // so this equality based remove provably drops this domainMap and no other.
+        // Unlike removeDomain(), this path clears no ChangeNumberIndexer state, and a later
+        // creation broadcasts addDomain() anew to multi domain cursors which already incorporated
+        // the domain: MultiDomainDBCursor discards such an announcement when it incorporates new
+        // cursors, so no second cursor is opened over the domain.
+        if (domainMap.isEmpty())
+        {
+          domainToReplicaDBs.remove(baseDN, domainMap);
+        }
+      }
     }
   }
 
@@ -326,6 +356,20 @@ FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final Replicatio
     return new FileReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv);
   }
 
+  /**
+   * Returns the map of replica DBs per domain.
+   * <p>
+   * Package private, for tests only: they both observe which domain maps this changelog holds and
+   * mutate the map to drive race interleavings, so this getter intentionally returns the live
+   * internal map, not a copy or an unmodifiable view.
+   *
+   * @return the live map of replica DBs per domain
+   */
+  ConcurrentMap<DN, ConcurrentMap<Integer, FileReplicaDB>> getDomainToReplicaDBs()
+  {
+    return domainToReplicaDBs;
+  }
+
   @Override
   public void initializeDB() throws ChangelogException
   {
@@ -403,13 +447,23 @@ public void shutdownDB() throws ChangelogException
       firstException = e;
     }
 
-    for (Iterator<ConcurrentMap<Integer, FileReplicaDB>> it =
-        this.domainToReplicaDBs.values().iterator(); it.hasNext();)
+    for (Iterator<Map.Entry<DN, ConcurrentMap<Integer, FileReplicaDB>>> it =
+        this.domainToReplicaDBs.entrySet().iterator(); it.hasNext();)
     {
-      final ConcurrentMap<Integer, FileReplicaDB> domainMap = it.next();
+      final Map.Entry<DN, ConcurrentMap<Integer, FileReplicaDB>> entry = it.next();
+      final ConcurrentMap<Integer, FileReplicaDB> domainMap = entry.getValue();
       synchronized (domainMap)
       {
-        it.remove();
+        // Follow the removal protocol documented on domainToReplicaDBs: unmap only the domainMap
+        // instance the monitor was taken on. The check is identity based, like removeDomain()'s:
+        // an equality based remove could still drop a map whose monitor is not held, since two
+        // empty maps are equal and removeDomain() plus a fresh creation may have swapped one for
+        // another since this iterator read its entry. Replica DBs another remover already visited
+        // are shut down again below, which is harmless: shutdown is a no-op the second time.
+        if (domainToReplicaDBs.get(entry.getKey()) == domainMap)
+        {
+          domainToReplicaDBs.remove(entry.getKey());
+        }
         for (FileReplicaDB replicaDB : domainMap.values())
         {
           replicaDB.shutdown();

opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java

@@ -12,12 +12,14 @@
  * information: "Portions Copyright [year] [name of copyright owner]".
  *
  * Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.replication.server.changelog.file;
 
 import java.util.Iterator;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ConcurrentSkipListSet;
 
 import net.jcip.annotations.NotThreadSafe;
 
@@ -34,6 +36,18 @@ public class MultiDomainDBCursor extends CompositeDBCursor<DN>
 {
   private final ReplicationDomainDB domainDB;
   private final ConcurrentSkipListMap<DN, ServerState> newDomains = new ConcurrentSkipListMap<>();
+  /**
+   * The domains this cursor already iterates over, so that a second announcement of a domain is
+   * discarded at incorporation: a second cursor over the same domain would either leak unclosed
+   * - the cursor tree of {@link CompositeDBCursor} collapses cursors comparing equal - or
+   * deliver every change twice. A domain may be announced again after the empty domainMap of a
+   * failed replica DB creation was dropped, while this cursor still iterates over it.
+   * <p>
+   * Added to and removed from on the thread iterating this cursor - {@link #incorporateNewCursors()}
+   * and {@link #removeDomain(DN)} run on it - read by the threads announcing domains, and
+   * cleared by {@link #close()}, which an ending ECL session may call from another thread.
+   */
+  private final ConcurrentSkipListSet<DN> incorporatedDomains = new ConcurrentSkipListSet<>();
   private final CursorOptions options;
 
   /**
@@ -52,6 +66,10 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt
   /**
    * Adds a replication domain for this cursor to iterate over. Added cursors
    * will be created and iterated over on the next call to {@link #next()}.
+   * <p>
+   * Announcing a domain this cursor already iterates over has no effect: the announcement is
+   * discarded when cursors are incorporated, and new replica DBs of such a domain reach it
+   * through {@link DomainDBCursor#addReplicaDB(int, org.opends.server.replication.common.CSN)}.
    *
    * @param baseDN
    *          the replication domain's baseDN
@@ -60,6 +78,9 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt
    */
   public void addDomain(DN baseDN, ServerState startAfterState)
   {
+    // incorporateNewCursors() discards announcements of domains this cursor already iterates
+    // over: checking incorporatedDomains here would be a check-then-act against removeDomain()
+    // on the cursor's thread, able to drop an announcement the removal no longer covers
     newDomains.put(baseDN, startAfterState != null ? startAfterState : new ServerState());
   }
 
@@ -73,8 +94,14 @@ protected void incorporateNewCursors() throws ChangelogException
       final Entry<DN, ServerState> entry = iter.next();
       final DN baseDN = entry.getKey();
       final ServerState serverState = entry.getValue();
-      final DBCursor<UpdateMsg> domainDBCursor = domainDB.getCursorFrom(baseDN, serverState, options);
-      addCursor(domainDBCursor, baseDN);
+      // discard the announcement of a domain this cursor already iterates over: this is the only
+      // thread adding to and removing from incorporatedDomains, so the check cannot race them
+      if (!incorporatedDomains.contains(baseDN))
+      {
+        final DBCursor<UpdateMsg> domainDBCursor = domainDB.getCursorFrom(baseDN, serverState, options);
+        addCursor(domainDBCursor, baseDN);
+        incorporatedDomains.add(baseDN);
+      }
       iter.remove();
     }
   }
@@ -90,6 +117,7 @@ protected void incorporateNewCursors() throws ChangelogException
   public void removeDomain(DN baseDN)
   {
     removeCursor(baseDN);
+    incorporatedDomains.remove(baseDN);
   }
 
   /** {@inheritDoc} */
@@ -99,6 +127,7 @@ public void close()
     super.close();
     domainDB.unregisterCursor(this);
     newDomains.clear();
+    incorporatedDomains.clear();
   }
 
 }

opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/ECLMultiDomainDBCursorTest.java

@@ -12,10 +12,13 @@
  * information: "Portions Copyright [year] [name of copyright owner]".
  *
  * Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.replication.server.changelog.file;
 
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 import org.forgerock.opendj.ldap.DN;
@@ -45,6 +48,8 @@ public class ECLMultiDomainDBCursorTest extends DirectoryServerTestCase
   private MultiDomainDBCursor multiDomainCursor;
   private ECLMultiDomainDBCursor eclCursor;
   private final Set<DN> eclEnabledDomains = new HashSet<>();
+  /** The long-lived cursor of each domain announced to {@link #multiDomainCursor}. */
+  private final Map<DN, SequentialDBCursor> domainCursors = new HashMap<>();
   private ECLEnabledDomainPredicate predicate = new ECLEnabledDomainPredicate()
   {
     @Override
@@ -61,6 +66,9 @@ public void setup() throws Exception
     options = new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY);
     multiDomainCursor = new MultiDomainDBCursor(domainDB, options);
     eclCursor = new ECLMultiDomainDBCursor(predicate, multiDomainCursor);
+    // one test class instance runs all the methods: the domains announced to the previous
+    // method's multiDomainCursor must not be mistaken for domains this one iterates over
+    domainCursors.clear();
   }
 
   @AfterMethod
@@ -185,6 +193,18 @@ private void assertMessagesInOrder(DN baseDN, UpdateMsg msg1, UpdateMsg msg2) th
 
   private void addDomainCursorToCursor(DN baseDN, SequentialDBCursor cursor) throws ChangelogException
   {
+    final SequentialDBCursor existing = domainCursors.get(baseDN);
+    if (existing != null)
+    {
+      // already known to the cursor: its long-lived per-domain cursor receives the new changes,
+      // exactly as DomainDBCursor.addReplicaDB() does in production
+      for (UpdateMsg msg : cursor.drain())
+      {
+        existing.add(msg);
+      }
+      return;
+    }
+    domainCursors.put(baseDN, cursor);
     final ServerState state = new ServerState();
     when(domainDB.getCursorFrom(baseDN, state, options)).thenReturn(cursor);
     multiDomainCursor.addDomain(baseDN, state);