mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
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; 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 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 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();
