Improve ReplicationDomainTest stability

opendj-server-legacy/pom.xml

@@ -1274,6 +1274,8 @@
                     <org.opends.test.suppressOutput>true</org.opends.test.suppressOutput>
                     <org.opends.test.pauseOnFailure>false</org.opends.test.pauseOnFailure>
                     <org.opends.test.copyClassesToTestPackage>false</org.opends.test.copyClassesToTestPackage>
+                    <org.opends.test.timeout>600000</org.opends.test.timeout><!--15 mins-->
+                    <org.opends.test.trace.pattern>(org\.opends\.server\.replication\.service\..*)|(org\.opends\.server\.replication\.GenerationIdTest)|(org\.opends\.server\.types.\HostPortTest)</org.opends.test.trace.pattern>
                   </systemPropertyVariables>
                   <argLine>@{argLine}</argLine>
 	          	  <reuseForks>false</reuseForks>

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

@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2010 Sun Microsystems, Inc.
  * Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2023-2025 3A Systems LLC.
  * Portions Copyright 2025 Wren Security.
  */
 package org.opends.server.replication.service;
@@ -2665,14 +2666,16 @@ public void stop()
     {
       debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
     }
+    synchronized (startStopLock) {
+      if (shutdown) {
+        return;
+      }
+    }
+
+    domain.publishReplicaOfflineMsg();
 
     synchronized (startStopLock)
     {
-      if (shutdown)
-      {
-        return;
-      }
-      domain.publishReplicaOfflineMsg();
       shutdown = true;
       setConnectedRS(ConnectedRS.stopped());
       stopRSHeartBeatMonitoring();

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

@@ -13,6 +13,7 @@
  *
  * Copyright 2008-2010 Sun Microsystems, Inc.
  * Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2025 3A Systems LLC.
  */
 package org.opends.server.replication.service;
 
@@ -39,6 +40,9 @@
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -372,6 +376,11 @@ public ECLIncludes addIncludedAttributes(int serverId,
    */
   protected volatile long generationId;
 
+  /**
+   * Thread pool for export thread tasks
+   */
+  private final ExecutorService exportThreadPool = Executors.newCachedThreadPool();
+
   /**
    * Returns the {@link CSNGenerator} that will be used to
    * generate {@link CSN} for this domain.
@@ -879,7 +888,10 @@ else if (msg instanceof InitializeRcvAckMsg)
         // Do this work in a thread to allow replay thread continue working
         ExportThread exportThread = new ExportThread(
             initReqMsg.getSenderID(), initReqMsg.getInitWindow());
-        exportThread.start();
+        exportThreadPool.execute(() -> {
+          Thread.currentThread().setName(exportThread.getName());
+          exportThread.run();
+        });
       }
     }
 
@@ -1075,6 +1087,9 @@ public void run()
         This server is not the initiator of the export so there is
         nothing more to do locally.
         */
+        if (logger.isTraceEnabled()) {
+          logger.trace(LocalizableMessage.raw("[IE] got exception" + getName()), de);
+        }
       }
 
       if (logger.isTraceEnabled())
@@ -3016,6 +3031,13 @@ public void disableService()
       {
         broker.stop();
       }
+      try {
+        exportThreadPool.shutdown();
+        boolean timedOut = exportThreadPool.awaitTermination(100, TimeUnit.SECONDS);
+        logger.info(LocalizableMessage.raw("export pool termination timed out: " + timedOut));
+      } catch (InterruptedException e) {
+        // Give up waiting.
+      }
 
       // Stop the listener thread
       if (listenerThread != null)

opendj-server-legacy/src/main/java/org/opends/server/types/HostPort.java

@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2008 Sun Microsystems, Inc.
  * Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2025 3A Systems LLC.
  */
 package org.opends.server.types;
 
@@ -459,6 +460,9 @@ public boolean isEquivalentTo(final HostPort other)
       // Get and compare ports of RS1 and RS2
       if (getPort() != other.getPort())
       {
+        if(logger.isTraceEnabled()) {
+          logger.trace("port and host does not match " + this + "; " + other);
+        }
         return false;
       }
 
@@ -475,6 +479,9 @@ public boolean isEquivalentTo(final HostPort other)
       }
       else if (thisAddresses == null || otherAddresses == null)
       {
+        if(logger.isTraceEnabled()) {
+          logger.trace("port and host does not match: " + this + "=" + thisAddresses + "; " + other + "=" + otherAddresses);
+        }
         // One local address and one non-local.
         return false;
       }
@@ -490,10 +497,17 @@ else if (thisAddresses == null || otherAddresses == null)
           }
         }
       }
+      if(logger.isTraceEnabled()) {
+        logger.trace("port and host does not match: " + this + "=" + thisAddresses + "; " + other + "=" + otherAddresses);
+      }
+
       return false;
     }
     catch (UnknownHostException ex)
     {
+      if(logger.isTraceEnabled()) {
+        logger.traceException(ex, "got exception when resolving hosts: " + this + " and " + other );
+      }
       // Unknown RS: should not happen
       return false;
     }

opendj-server-legacy/src/test/java/org/opends/server/DirectoryServerTestCase.java

@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2008 Sun Microsystems, Inc.
  * Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2025 3A Systems LLC.
  */
 package org.opends.server;
 
@@ -48,7 +49,7 @@ public final void suppressOutput() {
     TestCaseUtils.suppressOutput();
   }
 
-  @AfterSuite
+  @AfterSuite(alwaysRun = true)
   public final void shutdownServer() {
     TestCaseUtils.shutdownServer(LocalizableMessage.raw("The current test suite has finished."));
     TestCaseUtils.unsupressOutput();

opendj-server-legacy/src/test/java/org/opends/server/TestCaseUtils.java

@@ -45,11 +45,12 @@
 import java.io.OutputStream;
 import java.io.PrintStream;
 import java.io.StringReader;
+import java.lang.management.LockInfo;
 import java.lang.management.ManagementFactory;
+import java.lang.management.MonitorInfo;
 import java.lang.management.ThreadInfo;
 import java.lang.management.ThreadMXBean;
 import java.net.BindException;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
@@ -542,6 +543,9 @@ private static void setupLoggers()
     ErrorLogger.getInstance().addLogPublisher(
             (ErrorLogPublisher) getServerStartupTextErrorPublisher(ERROR_TEXT_WRITER));
 
+  }
+
+  public static void setupTrace() {
     DebugLogger.getInstance().addPublisherIfRequired(DEBUG_TEXT_WRITER);
   }
 
@@ -742,11 +746,20 @@ public synchronized static ServerSocket bindFreePort() throws IOException
   {
 	  for (; port > 1024;)
 	  {
+         ServerSocket res=null;
 	     try
 	     {
-	       return bindPort(port--);
+           res=bindPort(port--);
+	       return res;
 	     }
-	     catch (BindException e){}		
+	     catch (BindException e){
+             if (res!=null) {
+                 try {
+                     res.close();
+                 } catch (IOException ex) {}
+             }
+             res=null;
+         }
 	  }
 	  throw new BindException("Unable to bind to a free port");
   }
@@ -1927,20 +1940,71 @@ public static void generateThreadDump(String id)
   public static String generateThreadDump() {
     final StringBuilder dump = new StringBuilder();
     final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
-    final ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 100);
+    ThreadInfo[] threadInfos = threadMXBean.dumpAllThreads(true, true);
     for (ThreadInfo threadInfo : threadInfos) {
-        dump.append('"');
-        dump.append(threadInfo.getThreadName());
-        dump.append("\" ");
-        final Thread.State state = threadInfo.getThreadState();
-        dump.append("\n   java.lang.Thread.State: ");
-        dump.append(state);
-        final StackTraceElement[] stackTraceElements = threadInfo.getStackTrace();
-        for (final StackTraceElement stackTraceElement : stackTraceElements) {
-            dump.append("\n        at ");
-            dump.append(stackTraceElement);
+      dump.append("\"" + threadInfo.getThreadName() + "\"" +
+              (threadInfo.isDaemon() ? " daemon" : "") +
+              " prio=" + threadInfo.getPriority() +
+              " Id=" + threadInfo.getThreadId() + " " +
+              threadInfo.getThreadState());
+      if (threadInfo.getLockName() != null) {
+        dump.append(" on " + threadInfo.getLockName());
+      }
+      if (threadInfo.getLockOwnerName() != null) {
+        dump.append(" owned by \"" + threadInfo.getLockOwnerName() +
+                "\" Id=" + threadInfo.getLockOwnerId());
+      }
+      if (threadInfo.isSuspended()) {
+        dump.append(" (suspended)");
+      }
+      if (threadInfo.isInNative()) {
+        dump.append(" (in native)");
+      }
+      dump.append('\n');
+      StackTraceElement[] stackTrace = threadInfo.getStackTrace();
+      int i = 0;
+      for (; i < stackTrace.length; i++) {
+        StackTraceElement ste = stackTrace[i];
+        dump.append("\tat " + ste.toString());
+        dump.append('\n');
+        if (i == 0 && threadInfo.getLockInfo() != null) {
+          Thread.State ts = threadInfo.getThreadState();
+          switch (ts) {
+            case BLOCKED:
+              dump.append("\t-  blocked on " + threadInfo.getLockInfo());
+              dump.append('\n');
+              break;
+            case WAITING:
+            case TIMED_WAITING:
+              dump.append("\t-  waiting on " + threadInfo.getLockInfo());
+              dump.append('\n');
+              break;
+            default:
+          }
         }
-        dump.append("\n\n");
+
+        for (MonitorInfo mi : threadInfo.getLockedMonitors()) {
+          if (mi.getLockedStackDepth() == i) {
+            dump.append("\t-  locked " + mi);
+            dump.append('\n');
+          }
+        }
+      }
+      if (i < stackTrace.length) {
+        dump.append("\t...");
+        dump.append('\n');
+      }
+
+      LockInfo[] locks = threadInfo.getLockedSynchronizers();
+      if (locks.length > 0) {
+        dump.append("\n\tNumber of locked synchronizers = " + locks.length);
+        dump.append('\n');
+        for (LockInfo li : locks) {
+          dump.append("\t- " + li);
+          dump.append('\n');
+        }
+      }
+      dump.append('\n');
     }
     return dump.toString();
   }

opendj-server-legacy/src/test/java/org/opends/server/TestListener.java

@@ -13,7 +13,7 @@
  *
  * Copyright 2008 Sun Microsystems, Inc.
  * Portions Copyright 2013-2016 ForgeRock AS.
- * Portions Copyright 2023 3A Systems, LLC.
+ * Portions Copyright 2023-2025 3A Systems, LLC.
  */
 package org.opends.server;
 
@@ -155,11 +155,26 @@ public TestListener() throws Exception {
   @Override
   public void onStart(ITestContext testContext) {
     super.onStart(testContext);
-    
+
     if (testContext.getAllTestMethods().length>0) {
     	TestCaseUtils.setTestName(testContext.getAllTestMethods()[0].getInstance().getClass().getName());
     }
-
+    long testTimeout = 0;
+    try {
+      testTimeout = Long.parseLong(System.getProperty("org.opends.test.timeout", "0"));
+    } catch (NumberFormatException ignored) {}
+
+    for (int i = 0; i < testContext.getAllTestMethods().length; i++) {
+      testContext.getAllTestMethods()[i].setTimeOut(testTimeout);
+    }
+    if(System.getProperty("org.opends.test.trace.pattern") != null && testContext.getAllTestMethods().length > 0) {
+      String tracePattern = System.getProperty("org.opends.test.trace.pattern");
+      if(testContext.getAllTestMethods()[0].getInstance().getClass().getName().matches(tracePattern)) {
+        System.setProperty("org.opends.server.debug.target.1", "_global:enabled");
+        TestCaseUtils.setupTrace();
+      }
+    }
+
     // Delete the previous report if it's there.
     new File(testContext.getOutputDirectory(), REPORT_FILE_NAME).delete();
   }
@@ -179,6 +194,7 @@ public void onFinish(ITestContext testContext) {
 			originalSystemErr.println("check state: "+paths.unitRoot);
 		}
 	}
+    System.clearProperty("org.opends.server.debug.target.1");
   }
 
   @Override
@@ -335,6 +351,7 @@ private void writeReportToScreen(File reportFile) {
   @Override
   public void onTestStart(ITestResult tr) {
     super.onTestStart(tr);
+    originalSystemOut.println("-- Executing test: " +  tr.getMethod());
 
     enforceTestClassTypeAndAnnotations(tr);
     checkForInterleavedBetweenClasses(tr);
@@ -358,6 +375,11 @@ public void onTestSuccess(ITestResult tr) {
   public void onTestFailure(ITestResult tr) {
     super.onTestFailure(tr);
     reportTestFailed(tr);
+    printThreadDump();
+  }
+
+  private void printThreadDump() {
+    originalSystemErr.println(TestCaseUtils.generateThreadDump());
   }
 
   private void reportTestFailed(ITestResult tr)

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

@@ -921,6 +921,8 @@ public void testMultiRS() throws Exception
       replServer3 = createReplicationServer(replServerId3, true, testCase);
 
       connectServer1ToReplServer(replServer1);
+      Thread.sleep(2000); //wait for all RS handshakes to complete
+
       debugInfo("Expect genId are set in all replServers.");
       waitForStableGenerationId(EMPTY_DN_GENID);
       disconnectFromReplServer(replServer1);

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

@@ -13,6 +13,7 @@
  *
  * Copyright 2008-2010 Sun Microsystems, Inc.
  * Portions Copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2025 3A Systems LLC.
  */
 package org.opends.server.replication;
 
@@ -234,6 +235,7 @@ public void pushSchemaFilesChange() throws Exception
         @Override
         public void call() throws Exception
         {
+          assertTrue(schemaFile.exists());
           String fileStr = readAsString(schemaFile);
           assertTrue(fileStr.contains(stateStr), "The Schema persistentState (CSN:" + stateStr
               + ") has not been saved to " + schemaFile + " : " + fileStr);