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 68e2999e26..cc05d77b1f 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 @@ protected void initializeRemote(int serverToInitialize, // subsequent total update as a simultaneous import/export. final Map 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 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; 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 a6457a780b..367a21191f 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 @@ 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 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 c5215b013a..4703794cb0 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 @@ 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. */ 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 27e11d85c2..fd03a26bbd 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 @@ public void exportAndImportAcross2ReplServers() throws Exception } } + /** + * 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 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 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();