[#841] Fix flaky InitOnLineTest: notify the requester when a remotely requested export cannot start

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java

@@ -1469,30 +1469,78 @@ protected void initializeRemote(int serverToInitialize,
     // subsequent total update as a simultaneous import/export.
     final Map<Integer, DSInfo> replicaInfos = getReplicaInfos();
     final DSInfo targetDsi;
-    if (serverToInitialize == RoutableMsg.ALL_SERVERS)
+    final ImportExportContext ieCtx;
+    final long entryCount;
+    try
     {
-      if (replicaInfos.isEmpty())
+      if (serverToInitialize == RoutableMsg.ALL_SERVERS)
       {
-        throw new DirectoryException(UNWILLING_TO_PERFORM,
-            ERR_FULL_UPDATE_NO_REMOTES.get(getBaseDN(), getServerId()));
+        if (replicaInfos.isEmpty())
+        {
+          throw new DirectoryException(UNWILLING_TO_PERFORM,
+              ERR_FULL_UPDATE_NO_REMOTES.get(getBaseDN(), getServerId()));
+        }
+        targetDsi = null;
       }
-      targetDsi = null;
+      else
+      {
+        targetDsi = getDsInfoOrNull(replicaInfos.values(), serverToInitialize);
+        if (targetDsi == null)
+        {
+          throw new DirectoryException(UNWILLING_TO_PERFORM,
+              ERR_FULL_UPDATE_MISSING_REMOTE.get(getBaseDN(), getServerId(), serverToInitialize));
+        }
+      }
+
+      // countEntries() would otherwise first be called by
+      // initializeRemote(ieCtx, ...) outside the region that reports the
+      // failure to the requester: probe it here so a backend that cannot be
+      // exported is notified like any other rejection.
+      entryCount = countEntries();
+
+      ieCtx = acquireIEContext(false);
     }
-    else
+    catch (DirectoryException de)
     {
-      targetDsi = getDsInfoOrNull(replicaInfos.values(), serverToInitialize);
-      if (targetDsi == null)
+      if (initTask == null
+          && serverToInitialize != RoutableMsg.ALL_SERVERS
+          && serverRunningTheTask != getServerId())
       {
-        throw new DirectoryException(UNWILLING_TO_PERFORM,
-            ERR_FULL_UPDATE_MISSING_REMOTE.get(getBaseDN(), getServerId(), serverToInitialize));
+        /*
+        The export was requested by the remote server itself (the
+        ExportTask contract: no local task and the requester is the
+        target), which has acquired an import context and is now waiting
+        for the InitializeTargetMsg: without a reply it would wait forever
+        (e.g. when this request raced the topology propagation and the
+        requester is not in our replicas view yet). Best effort: the
+        requester may not even be routable in that very case - the
+        replication server then bounces the notification back as an
+        ErrorMsg(ERR_NO_REACHABLE_PEER) applied to whatever import/export
+        context is live here (ErrorMsg carries no correlation id) - and
+        when the session is down the requester detects the disconnection
+        instead.
+        */
+        logger.info(NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED,
+            getBaseDN(), getServerId(), serverToInitialize, de.getMessageObject());
+        try
+        {
+          if (broker.isConnected())
+          {
+            broker.publish(new ErrorMsg(serverToInitialize, de.getMessageObject()));
+          }
+        }
+        catch (Exception e)
+        {
+          // Ignore the failure raised while notifying the root failure
+        }
       }
+      throw de;
     }
 
-    final ImportExportContext ieCtx = acquireIEContext(false);
     try
     {
       initializeRemote(ieCtx, replicaInfos, targetDsi, serverToInitialize,
-          serverRunningTheTask, initTask, initWindow);
+          serverRunningTheTask, initTask, initWindow, entryCount);
     }
     finally
     {
@@ -1505,17 +1553,18 @@ protected void initializeRemote(int serverToInitialize,
 
   /**
    * Performs the remote initialization with the import/export context already
-   * acquired - and released - by the caller.
+   * acquired - and released - by the caller, which also counted the entries
+   * to export while validating the request.
    */
   private void initializeRemote(ImportExportContext ieCtx,
       Map<Integer, DSInfo> replicaInfos, DSInfo targetDsi,
       int serverToInitialize, int serverRunningTheTask, Task initTask,
-      int initWindow) throws DirectoryException
+      int initWindow, long entryCount) throws DirectoryException
   {
     if (serverToInitialize == RoutableMsg.ALL_SERVERS)
     {
       logger.info(NOTE_FULL_UPDATE_ENGAGED_FOR_REMOTE_START_ALL,
-          countEntries(), getBaseDN(), getServerId());
+          entryCount, getBaseDN(), getServerId());
 
       ieCtx.startList.addAll(replicaInfos.keySet());
 
@@ -1529,7 +1578,7 @@ private void initializeRemote(ImportExportContext ieCtx,
     }
     else
     {
-      logger.info(NOTE_FULL_UPDATE_ENGAGED_FOR_REMOTE_START, countEntries(),
+      logger.info(NOTE_FULL_UPDATE_ENGAGED_FOR_REMOTE_START, entryCount,
           getBaseDN(), getServerId(), serverToInitialize);
 
       ieCtx.startList.add(serverToInitialize);
@@ -1550,7 +1599,7 @@ private void initializeRemote(ImportExportContext ieCtx,
         {
           ieCtx.initializeTask = initTask;
         }
-        ieCtx.initializeCounters(countEntries());
+        ieCtx.initializeCounters(entryCount);
         ieCtx.msgCnt = 0;
         ieCtx.initNumLostConnections = broker.getNumLostConnections();
         ieCtx.initWindow = initWindow;

opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -608,3 +608,5 @@ ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING_304=Nothing accepted a connection on
  either by a socket bound to another address, or by a socket which does not accept connections
 ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port was released after the \
  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

opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java

@@ -52,7 +52,9 @@
 import org.opends.server.types.DirectoryException;
 import org.opends.server.types.Entry;
 import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 import static org.opends.messages.ReplicationMessages.*;
@@ -611,6 +613,10 @@ public void initializeExport() throws Exception
           server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
+      // The export is rejected when the InitializeRequestMsg arrives before
+      // the local domain sees DS2 in its topology view (issue #841)
+      waitForRemoteReplicas(server2ID);
+
       InitializeRequestMsg initMsg = new InitializeRequestMsg(baseDN, server2ID, server1ID, 100);
       server2.publish(initMsg);
 
@@ -1074,14 +1080,27 @@ public void initializeTargetUnknownRemote() throws Exception
   private void waitForInitializeTargetMsg(String testCase,
       ReplicationBroker server) throws Exception
   {
-    ReplicationMsg msgrcv;
-    do
+    // Fail fast when the initialization is lost or failed: looping until the
+    // TestNG method timeout would leave the replication servers (and their
+    // ports) running for the remaining tests of the class (issue #841).
+    final long deadline = System.currentTimeMillis() + 60000;
+    while (true)
     {
-      msgrcv = server.receive();
+      ReplicationMsg msgrcv = server.receive();
       log(testCase + " " + server.getServerId() + " receives " + msgrcv);
+      if (msgrcv instanceof InitializeTargetMsg)
+      {
+        return;
+      }
+      if (msgrcv == null || msgrcv instanceof ErrorMsg)
+      {
+        fail(testCase + ": waiting for InitializeTargetMsg, received " + msgrcv);
+      }
+      if (System.currentTimeMillis() > deadline)
+      {
+        fail(testCase + ": no InitializeTargetMsg received within 60s, last received " + msgrcv);
+      }
     }
-    while (!(msgrcv instanceof InitializeTargetMsg));
-    Assertions.assertThat(msgrcv).isInstanceOf(InitializeTargetMsg.class);
   }
 
   @Test(enabled=true)
@@ -1123,6 +1142,13 @@ public void initializeExportMultiSS() throws Exception
           10000, replServer1.getGenerationId(baseDN));
       }
 
+      // Wait for the local domain to see DS3 in its topology view before S3
+      // requests the initialization: the InitializeRequestMsg can outrun the
+      // TopologyMsg propagation (RS3 -> RS1 -> DS1), in which case the export
+      // is rejected with "the remote directory server DS(3) is unknown" and
+      // S3 never receives the InitializeTargetMsg (issue #841).
+      waitForRemoteReplicas(server3ID);
+
       // S3 sends init request
       log(testCase + " server 3 Will send reqinit to " + server1ID);
       InitializeRequestMsg initMsg = new InitializeRequestMsg(baseDN, server3ID, server1ID, 100);
@@ -1359,6 +1385,14 @@ private void waitForRemoteReplicas(Integer... serverIds) throws Exception
 
   private void afterTest(String testCase) throws Exception
   {
+    if (releasedByAfterMethod)
+    {
+      // this is the abandoned thread of a timed out test method, unblocked by
+      // releaseLeakedReplicationServers: the shared state was already
+      // neutralised and cleaned on the main thread, running the cleanup below
+      // concurrently would wreck the currently running test method
+      return;
+    }
     // Check that the domain has completed the import/export task.
     boolean ieStillRunning = false;
     if (replDomain != null)
@@ -1402,6 +1436,71 @@ private void afterTest(String testCase) throws Exception
     assertFalse(ieStillRunning, "ReplicationDomain: Import/Export is not expected to be running");
   }
 
+  /**
+   * Set when releaseLeakedReplicationServers cleaned up after a timed out
+   * test method: closing the leaked sessions unblocks the abandoned test
+   * thread, whose own finally{afterTest()} must then become a no-op instead
+   * of cleaning up the next test method. Written on the main thread before
+   * anything can wake the abandoned thread, read on afterTest's first line.
+   */
+  private volatile boolean releasedByAfterMethod;
+
+  @BeforeMethod(alwaysRun = true)
+  public void resetReleasedByAfterMethod()
+  {
+    releasedByAfterMethod = false;
+  }
+
+  /**
+   * Releases what a timed out test method left behind: TestNG abandons the
+   * test thread on a thread timeout, the finally block of the test never
+   * completes, and the domain config entry and the listen ports (cached in
+   * replServerPort) would otherwise poison the remaining tests of the class
+   * (issue #841). Successful tests clean up in afterTest, which nulls every
+   * field checked here. Runs on the main thread - TestNG still runs
+   * configuration methods after a thread timeout.
+   */
+  @AfterMethod(alwaysRun = true)
+  public void releaseLeakedReplicationServers()
+  {
+    if (replServer1 == null && replServer2 == null && replServer3 == null
+        && server2 == null && server3 == null && replDomain == null)
+    {
+      // the test method cleaned up after itself
+      return;
+    }
+    log("Releasing the replication servers leaked by a timed out test");
+    // Neutralise the shared state *before* anything can wake the abandoned
+    // test thread: stopping its broker unblocks receive(), and its own
+    // finally{afterTest()} would otherwise clean up the *next* test.
+    releasedByAfterMethod = true;
+    final ReplicationBroker b2 = server2, b3 = server3;
+    final ReplicationServer rs1 = replServer1, rs2 = replServer2, rs3 = replServer3;
+    server2 = server3 = null;
+    replServer1 = replServer2 = replServer3 = null;
+    replDomain = null;
+    Arrays.fill(replServerPort, 0);
+    // best effort: throwing from an @AfterMethod would skip the rest of the
+    // class (configfailurepolicy=skip), which is worse than the leak
+    try
+    {
+      super.cleanConfigEntries();
+    }
+    catch (Throwable t)
+    {
+      log("Failed to remove the leaked domain configuration: " + t);
+    }
+    try
+    {
+      stop(b2, b3);
+      remove(rs1, rs2, rs3);
+    }
+    catch (Throwable t)
+    {
+      log("Failed to release the leaked replication servers: " + t);
+    }
+  }
+
   /**
    * Clean up the environment.
    */