From 8dcf971d0d02b2d86ca00e1906d14827fc6f31cb Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 10:18:00 +0300 Subject: [PATCH 1/3] [#841] Fix flaky InitOnLineTest: notify the requester when a remotely requested export cannot start When a total update is requested by a remote replica (InitializeRequestMsg, no local task), a failure before the export starts - the requester missing from the replicas view after racing the topology propagation, or an import/export already in progress - was thrown without notifying anyone: the requester kept waiting for an InitializeTargetMsg that never came. Send the failure back as an ErrorMsg (best effort), as the routing failure paths already do. On the test side, initializeExportMultiSS published the InitializeRequestMsg right after connecting the brokers, so it could race the TopologyMsg propagation (RS3 -> RS1 -> DS1) and hang for the whole 600 s method timeout in waitForInitializeTargetMsg: the loop ignored ErrorMsg and null, and the 10 s broker soTimeout never fired because the replication server publishes a MonitorMsg every 3 s which the broker consumes internally. The timed-out thread then skipped its finally cleanup and left the replication servers bound to their ports, cascading into the BindException in initializeImport. - wait for the local domain to see the requester in its topology view before publishing the InitializeRequestMsg (initializeExport, initializeExportMultiSS) - make waitForInitializeTargetMsg fail fast on ErrorMsg, on a closed connection and after 60 s - cover the notification with ReplicationDomainTest.remotelyRequestedExportFailureNotifiesRequester --- .../service/ReplicationDomain.java | 50 +++++++-- .../server/replication/InitOnLineTest.java | 34 +++++- .../service/ReplicationDomainTest.java | 104 +++++++++++++++++- 3 files changed, 171 insertions(+), 17 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 68e2999e26..1133f42b8c 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,26 +1469,54 @@ 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; + try { - if (replicaInfos.isEmpty()) + if (serverToInitialize == RoutableMsg.ALL_SERVERS) + { + if (replicaInfos.isEmpty()) + { + throw new DirectoryException(UNWILLING_TO_PERFORM, + ERR_FULL_UPDATE_NO_REMOTES.get(getBaseDN(), getServerId())); + } + targetDsi = null; + } + else { - throw new DirectoryException(UNWILLING_TO_PERFORM, - ERR_FULL_UPDATE_NO_REMOTES.get(getBaseDN(), getServerId())); + targetDsi = getDsInfoOrNull(replicaInfos.values(), serverToInitialize); + if (targetDsi == null) + { + throw new DirectoryException(UNWILLING_TO_PERFORM, + ERR_FULL_UPDATE_MISSING_REMOTE.get(getBaseDN(), getServerId(), serverToInitialize)); + } } - targetDsi = null; + + ieCtx = acquireIEContext(false); } - else + catch (DirectoryException de) { - targetDsi = getDsInfoOrNull(replicaInfos.values(), serverToInitialize); - if (targetDsi == null) + if (initTask == null) { - throw new DirectoryException(UNWILLING_TO_PERFORM, - ERR_FULL_UPDATE_MISSING_REMOTE.get(getBaseDN(), getServerId(), serverToInitialize)); + /* + The export was requested by the remote server itself, 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. + */ + try + { + 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, 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..476b38954b 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 @@ -611,6 +611,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 +1078,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 +1140,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); 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..4499a0503c 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,106 @@ 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); + if (firstExport != null) + { + // losing broker2 empties the exporter start list and ends the export + firstExport.join(30000); + assertFalse(firstExport.isAlive(), "the first export did not terminate"); + } + disable(domain1); + remove(replServer); + } + } + private String buildExportedData(final int ENTRYCOUNT) { final StringBuilder sb = new StringBuilder(); From 6ec477300647967edeae1f55cf5305b644814930 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 15:30:44 +0300 Subject: [PATCH 2/3] [#841] Address review feedback on the remotely requested export failure notification - probe countEntries() before acquiring the import/export context so a backend that cannot be exported is reported to the requester too - gate the ErrorMsg notification on the exact ExportThread contract to avoid a hypothetical ErrorMsg(ALL_SERVERS) fan-out from a local caller - skip the notification while the broker is disconnected and log the rejection on the exporter (new NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED) - InitOnLineTest: release the replication servers of a timed out test method in an @AfterMethod, outside the abandoned test thread - ReplicationDomainTest: assert after the cleanup in the finally block; cover the ERR_FULL_UPDATE_MISSING_REMOTE rejection --- .../service/ReplicationDomain.java | 26 +++++++--- .../opends/messages/replication.properties | 2 + .../server/replication/InitOnLineTest.java | 26 ++++++++++ .../service/ReplicationDomainTest.java | 52 ++++++++++++++++++- 4 files changed, 99 insertions(+), 7 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 1133f42b8c..af32788eec 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 @@ -1491,23 +1491,37 @@ protected void initializeRemote(int serverToInitialize, } } + // countEntries() is 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. + countEntries(); + ieCtx = acquireIEContext(false); } catch (DirectoryException de) { - if (initTask == null) + if (initTask == null + && serverToInitialize != RoutableMsg.ALL_SERVERS + && serverRunningTheTask != getServerId()) { /* - The export was requested by the remote server itself, which has - acquired an import context and is now waiting for the - InitializeTargetMsg: without a reply it would wait forever + The export was requested by the remote server itself (the + ExportThread 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. + requester may not even be routable in that very case, and when the + session is down the requester detects the disconnection instead. */ + logger.info(NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED, + getServerId(), serverToInitialize, getBaseDN(), de.getMessageObject()); try { - broker.publish(new ErrorMsg(serverToInitialize, de.getMessageObject())); + if (broker.isConnected()) + { + broker.publish(new ErrorMsg(serverToInitialize, de.getMessageObject())); + } } catch (Exception e) { 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..aeb15ab9bf 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=This directory server DS(%d) cannot start the total update \ + requested by the remote directory server DS(%d) in domain "%s": %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 476b38954b..1a5bdf0705 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,6 +52,7 @@ 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.Test; @@ -1426,6 +1427,31 @@ private void afterTest(String testCase) throws Exception assertFalse(ieStillRunning, "ReplicationDomain: Import/Export is not expected to be running"); } + /** + * Releases the replication servers 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 listen ports would otherwise stay bound + * (and cached in replServerPort) for the remaining tests of the class + * (issue #841). Successful tests clean up in afterTest, which nulls every + * field checked here. + */ + @AfterMethod(alwaysRun = true) + public void releaseLeakedReplicationServers() throws Exception + { + if (replServer1 == null && replServer2 == null && replServer3 == null) + { + // the test method cleaned up after itself + return; + } + log("Releasing the replication servers leaked by a timed out test"); + stop(server2, server3); + server2 = server3 = null; + remove(replServer1, replServer2, replServer3); + replServer1 = replServer2 = replServer3 = null; + replDomain = null; + Arrays.fill(replServerPort, 0); + } + /** * 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 4499a0503c..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 @@ -547,14 +547,64 @@ public void remotelyRequestedExportFailureNotifiesRequester() throws Exception finally { stop(broker2); + boolean firstExportStillRunning = false; if (firstExport != null) { // losing broker2 empties the exporter start list and ends the export firstExport.join(30000); - assertFalse(firstExport.isAlive(), "the first export did not terminate"); + 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); } } From 516c039952d0714ed99999ff73366e67e69a17db Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 19:38:45 +0300 Subject: [PATCH 3/3] [#841] Make the timed-out-test net safe for the abandoned test thread - releaseLeakedReplicationServers neutralises the shared fields before closing the leaked sessions; afterTest early-returns on the releasedByAfterMethod flag (reset in @BeforeMethod), so the woken abandoned thread cannot clean up the next test method - the net also removes the leaked domain config entry and never throws: configfailurepolicy=skip would turn one cleanup failure into a whole-class skip - thread the entry count probed during validation into the private initializeRemote overload instead of re-counting three times - reword NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED to match its neighbours; document the ErrorMsg bounce path in the rejection comment; ExportThread -> ExportTask after the rename on master --- .../service/ReplicationDomain.java | 35 ++++++---- .../opends/messages/replication.properties | 4 +- .../server/replication/InitOnLineTest.java | 67 ++++++++++++++++--- 3 files changed, 81 insertions(+), 25 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 af32788eec..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 @@ -1470,6 +1470,7 @@ protected void initializeRemote(int serverToInitialize, final Map replicaInfos = getReplicaInfos(); final DSInfo targetDsi; final ImportExportContext ieCtx; + final long entryCount; try { if (serverToInitialize == RoutableMsg.ALL_SERVERS) @@ -1491,10 +1492,11 @@ protected void initializeRemote(int serverToInitialize, } } - // countEntries() is 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. - countEntries(); + // 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); } @@ -1506,16 +1508,20 @@ protected void initializeRemote(int serverToInitialize, { /* The export was requested by the remote server itself (the - ExportThread contract: no local task and the requester is 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, and when the - session is down the requester detects the disconnection instead. + 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, - getServerId(), serverToInitialize, getBaseDN(), de.getMessageObject()); + getBaseDN(), getServerId(), serverToInitialize, de.getMessageObject()); try { if (broker.isConnected()) @@ -1534,7 +1540,7 @@ The export was requested by the remote server itself (the try { initializeRemote(ieCtx, replicaInfos, targetDsi, serverToInitialize, - serverRunningTheTask, initTask, initWindow); + serverRunningTheTask, initTask, initWindow, entryCount); } finally { @@ -1547,17 +1553,18 @@ The export was requested by the remote server itself (the /** * 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()); @@ -1571,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); @@ -1592,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 aeb15ab9bf..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,5 +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=This directory server DS(%d) cannot start the total update \ - requested by the remote directory server DS(%d) in domain "%s": %s +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 1a5bdf0705..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 @@ -54,6 +54,7 @@ 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.*; @@ -1384,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) @@ -1428,28 +1437,68 @@ private void afterTest(String testCase) throws Exception } /** - * Releases the replication servers 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 listen ports would otherwise stay bound - * (and cached in replServerPort) for the remaining tests of the class + * 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. + * field checked here. Runs on the main thread - TestNG still runs + * configuration methods after a thread timeout. */ @AfterMethod(alwaysRun = true) - public void releaseLeakedReplicationServers() throws Exception + public void releaseLeakedReplicationServers() { - if (replServer1 == null && replServer2 == null && replServer3 == null) + 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"); - stop(server2, server3); + // 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; - remove(replServer1, replServer2, replServer3); 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); + } } /**