From 5bd63c0f087d5e0b8e44910301a69ad9481a71e0 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 05 Aug 2026 10:12:59 +0000
Subject: [PATCH] [#841] Fix flaky InitOnLineTest: notify the requester when a remotely requested export cannot start (#845)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 154 +++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java | 109 +++++++++++++++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 2
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 83 +++++++++++--
4 files changed, 325 insertions(+), 23 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 68e2999..cc05d77 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -1469,30 +1469,78 @@
// 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 @@
/**
* 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 @@
}
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 @@
{
ieCtx.initializeTask = initTask;
}
- ieCtx.initializeCounters(countEntries());
+ ieCtx.initializeCounters(entryCount);
ieCtx.msgCnt = 0;
ieCtx.initNumLostConnections = broker.getNumLostConnections();
ieCtx.initWindow = initWindow;
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 a6457a7..367a211 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -608,3 +608,5 @@
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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
index c5215b0..4703794 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
+++ b/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 @@
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 @@
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 @@
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 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)
@@ -1403,6 +1437,71 @@
}
/**
+ * 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.
*/
@AfterClass
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
index 27e11d8..fd03a26 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
@@ -13,7 +13,7 @@
*
* Copyright 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2025 3A Systems,LLC.
+ * Portions Copyright 2025-2026 3A Systems,LLC.
*/
package org.opends.server.replication.service;
@@ -40,6 +40,8 @@
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
+import org.opends.server.replication.protocol.ErrorMsg;
+import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
@@ -456,6 +458,156 @@
}
}
+ /**
+ * When an export requested by a remote replica cannot start (there is no
+ * local task reporting the failure), the requester keeps waiting for the
+ * InitializeTargetMsg: the exporter must send an ErrorMsg back, otherwise
+ * the requester waits forever (issue #841).
+ */
+ @Test(enabled=true)
+ public void remotelyRequestedExportFailureNotifiesRequester() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer = null;
+ FakeReplicationDomain domain1 = null;
+ ReplicationBroker broker2 = null;
+ Thread firstExport = null;
+
+ try
+ {
+ int replServerPort = TestCaseUtils.findFreePort();
+ replServer = createReplicationServer(11, replServerPort,
+ "remoteExportFailureNotifiesRequesterDb", 100);
+ SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);
+
+ String exportedData = buildExportedData(100);
+ domain1 = new FakeReplicationDomain(
+ testService, 1, servers, 0, exportedData, null, 100);
+
+ broker2 = openReplicationSession(testService, 2, 100, replServerPort,
+ 10000, domain1.getGenerationID());
+
+ final FakeReplicationDomain exporter = domain1;
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(() -> assertTrue(exporter.getReplicaInfos().containsKey(2),
+ "DS(2) is not known to the exporting domain"));
+
+ // Occupy the import/export context of the exporter: broker2 never
+ // enters the full update status, so this export stays in
+ // waitForRemoteStartOfInit until broker2 disconnects in the finally
+ firstExport = new Thread(() -> {
+ try
+ {
+ exporter.initializeRemote(2, 2, NO_INIT_TASK, 100);
+ }
+ catch (DirectoryException expected)
+ {
+ // broker2 never plays the importer role
+ }
+ });
+ firstExport.start();
+ TestTimer ieRunningTimer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ ieRunningTimer.repeatUntilSuccess(() -> assertTrue(exporter.ieRunning(),
+ "the first export did not acquire the import/export context"));
+
+ // A second remotely requested export is rejected...
+ try
+ {
+ domain1.initializeRemote(2, 2, NO_INIT_TASK, 100);
+ fail("Expected the simultaneous export to be rejected");
+ }
+ catch (DirectoryException expected)
+ {
+ assertEquals(expected.getMessageObject().toString(),
+ ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get().toString());
+ }
+
+ // ...and the requester is notified instead of waiting forever
+ final long deadline = System.currentTimeMillis() + 30000;
+ while (true)
+ {
+ ReplicationMsg msg = broker2.receive();
+ if (msg instanceof ErrorMsg)
+ {
+ assertEquals(((ErrorMsg) msg).getDetails().toString(),
+ ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get().toString());
+ break;
+ }
+ assertNotNull(msg, "connection closed while waiting for the ErrorMsg");
+ assertFalse(System.currentTimeMillis() > deadline,
+ "no ErrorMsg received within 30s, last received " + msg);
+ }
+ }
+ finally
+ {
+ stop(broker2);
+ boolean firstExportStillRunning = false;
+ if (firstExport != null)
+ {
+ // losing broker2 empties the exporter start list and ends the export
+ firstExport.join(30000);
+ firstExportStillRunning = firstExport.isAlive();
+ }
+ disable(domain1);
+ remove(replServer);
+ // asserted only after the cleanup above: failing before it would leak
+ // the domain and the replication server port into the following tests
+ assertFalse(firstExportStillRunning, "the first export did not terminate");
+ }
+ }
+
+ /**
+ * A total update requested by a remote replica that is not (yet) in the
+ * exporter's topology view - the request raced the TopologyMsg propagation,
+ * the actual issue #841 trigger - must be rejected without leaving the
+ * import/export context acquired. The ErrorMsg sent back cannot be asserted
+ * here: the replication server does not route messages to a replica it does
+ * not know about, and a requester that is connected yet still unknown to
+ * the exporter is exactly the race this rejection guards against.
+ */
+ @Test(enabled=true)
+ public void remotelyRequestedExportForUnknownReplicaIsRejected() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer = null;
+ FakeReplicationDomain domain1 = null;
+
+ try
+ {
+ int replServerPort = TestCaseUtils.findFreePort();
+ replServer = createReplicationServer(12, replServerPort,
+ "remoteExportUnknownReplicaDb", 100);
+ SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);
+
+ domain1 = new FakeReplicationDomain(
+ testService, 1, servers, 0, buildExportedData(10), null, 100);
+
+ try
+ {
+ domain1.initializeRemote(2, 2, NO_INIT_TASK, 100);
+ fail("Expected the export requested by an unknown replica to be rejected");
+ }
+ catch (DirectoryException expected)
+ {
+ assertEquals(expected.getMessageObject().toString(),
+ ERR_FULL_UPDATE_MISSING_REMOTE.get(testService, 1, 2).toString());
+ }
+ assertFalse(domain1.ieRunning(),
+ "the rejected export must not leave the import/export context acquired");
+ }
+ finally
+ {
+ disable(domain1);
+ remove(replServer);
+ }
+ }
+
private String buildExportedData(final int ENTRYCOUNT)
{
final StringBuilder sb = new StringBuilder();
--
Gitblit v1.10.0