/* * The contents of this file are subject to the terms of the Common Development and * Distribution License (the License). You may not use this file except in compliance with the * License. * * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the * specific language governing permission and limitations under the License. * * When distributing Covered Software, include this CDDL Header Notice in each file and include * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL * Header, with the fields enclosed by brackets [] replaced by your own identifying * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2008-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. * Portions Copyright 2025-2026 3A Systems, LLC. */ package org.opends.server.replication.service; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.TestCaseUtils.generateThreadDump; import static org.opends.server.util.CollectionUtils.*; import static org.testng.Assert.*; import java.net.SocketTimeoutException; import java.util.List; import java.util.Map; import java.util.SortedSet; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.i18n.slf4j.LocalizedLogger; import org.opends.server.TestCaseUtils; import org.opends.server.backends.task.Task; import org.opends.server.replication.ReplicationTestCase; import org.opends.server.replication.common.DSInfo; 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.ByteArrayBuilder; import org.opends.server.replication.protocol.ErrorMsg; import org.opends.server.replication.protocol.ProtocolVersion; 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; import org.opends.server.replication.server.ServerHandler; import org.opends.server.replication.service.ReplicationDomain.ImportExportContext; import org.forgerock.opendj.ldap.DN; import org.opends.server.tasks.InitializeTask; import org.opends.server.types.DirectoryException; import org.opends.server.util.TestTimer; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; /** * Test the Generic Replication Service. */ @SuppressWarnings("javadoc") public class ReplicationDomainTest extends ReplicationTestCase { private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); private static final Task NO_INIT_TASK = null; @DataProvider(name = "publishAndReceiveData") public Object[][] createpublishAndReceiveData() { return new Object[][] { {1, 2, 3, 4}, {1, 2, 1, 2}, {1, 2, 45891, 45672}, {45610, 45720, 1, 2}, {45610, 45720, 45891, 45672} }; } /** * Test that a ReplicationDomain is able to publish and receive UpdateMsg. * Also test the ReplicationDomain.resetReplicationLog() method. */ @Test(dataProvider = "publishAndReceiveData", enabled=true) public void publishAndReceive( int replServerID1, int replServerID2, int domain1ServerId, int domain2ServerId) throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; FakeReplicationDomain domain1 = null; FakeReplicationDomain domain2 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); int replServerPort1 = ports[0]; int replServerPort2 = ports[1]; replServer1 = createReplicationServer(replServerID1, replServerPort1, "ReplicationDomainTestDb1", 100, "localhost:" + replServerPort2); replServer2 = createReplicationServer(replServerID2, replServerPort2, "ReplicationDomainTestDb2", 100, "localhost:" + replServerPort1); SortedSet servers = newTreeSet("localhost:" + replServerPort1); BlockingQueue rcvQueue1 = new LinkedBlockingQueue<>(); domain1 = new FakeReplicationDomain( testService, domain1ServerId, servers, 100, 1000, rcvQueue1); SortedSet servers2 = newTreeSet("localhost:" + replServerPort2); BlockingQueue rcvQueue2 = new LinkedBlockingQueue<>(); domain2 = new FakeReplicationDomain( testService, domain2ServerId, servers2, 100, 1000, rcvQueue2); Thread.sleep(500); /* * Publish a message from domain1, * Check that domain2 receives it shortly after. */ byte[] test = {1, 2, 3 ,4, 0, 1, 2, 3, 4, 5}; publish(domain1, test); UpdateMsg rcvdMsg = rcvQueue2.poll(20, TimeUnit.SECONDS); assertNotNull(rcvdMsg); assertEquals(test, rcvdMsg.getPayload()); for (RSInfo replServerInfo : domain1.getRsInfos()) { // The generation Id of the remote should be 1 assertEquals(replServerInfo.getGenerationId(), 1, "Unexpected value of generationId in RSInfo for RS=" + replServerInfo); } for (DSInfo serverInfo : domain1.getReplicaInfos().values()) { assertEquals(serverInfo.getStatus(), ServerStatus.NORMAL_STATUS); } domain1.setGenerationID(2); domain1.resetReplicationLog(); Thread.sleep(500); for (RSInfo replServerInfo : domain1.getRsInfos()) { // The generation Id of the remote should now be 2 assertEquals(replServerInfo.getGenerationId(), 2, "Unexpected value of generationId in RSInfo for RS=" + replServerInfo); } int sleepTime = 50; while (true) { try { assertExpectedServerStatuses(domain1.getReplicaInfos(), domain1ServerId, domain2ServerId); assertExpectedServerStatuses(domain2.getReplicaInfos(), domain1ServerId, domain2ServerId); Map states1 = domain1.getReplicaStates(); ServerState state2 = states1.get(domain2ServerId); assertNotNull(state2, "getReplicaStates is not showing DS2"); Map states2 = domain2.getReplicaStates(); ServerState state1 = states2.get(domain1ServerId); assertNotNull(state1, "getReplicaStates is not showing DS1"); // if we reach this point all tests are OK break; } catch (AssertionError e) { if (sleepTime >= 30000) { throw e; } Thread.sleep(sleepTime); sleepTime *= 2; } } } finally { disable(domain1, domain2); remove(replServer1, replServer2); } } /** * Publish information to the Replication Service (not assured mode). * * @param msg The byte array containing the information that should * be sent to the remote entities. */ void publish(FakeReplicationDomain domain, byte[] msg) { UpdateMsg updateMsg; synchronized (this) { updateMsg = new UpdateMsg(domain.getGenerator().newCSN(), msg); // If assured replication is configured, // this will prepare blocking mechanism. // If assured replication is disabled, this returns immediately domain.prepareWaitForAckIfAssuredEnabled(updateMsg); domain.publish(updateMsg); } try { // If assured replication is enabled, // this will wait for the matching ack or time out. // If assured replication is disabled, this returns immediately domain.waitForAckIfAssuredEnabled(updateMsg); } catch (TimeoutException ex) { // This exception may only be raised if assured replication is enabled logger.info(NOTE_DS_ACK_TIMEOUT, domain.getBaseDN(), domain.getAssuredTimeout(), updateMsg); } } private void assertExpectedServerStatuses(Map dsInfos, int domain1ServerId, int domain2ServerId) { for (DSInfo serverInfo : dsInfos.values()) { if (serverInfo.getDsId() == domain2ServerId) { assertEquals(serverInfo.getStatus(), ServerStatus.BAD_GEN_ID_STATUS); } else { assertEquals(serverInfo.getDsId(), domain1ServerId); assertEquals(serverInfo.getStatus(), ServerStatus.NORMAL_STATUS); } } } /** * The session generation is the identity of a session: a replay thread which stopped the * session and let the lock go compares the generation it read then with the one it reads * when it comes back, and starts the session only if the two are the same. Both halves * of a restart have to move it, or a session stopped or started by something else in the * meantime would look like the one that thread stopped. */ @Test public void everyStopAndEveryStartOfTheSessionIsCounted() throws Exception { final DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain domain = null; try { final int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbGeneration", 100); domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1); final long live = sessionGenerationOf(domain); domain.disableService(); final long stopped = sessionGenerationOf(domain); assertNotEquals(stopped, live, "stopping the session left its generation where it was"); domain.enableService(); final long started = sessionGenerationOf(domain); assertNotEquals(started, stopped, "starting the session left its generation where it was"); } finally { disable(domain); remove(replServer); } } /** * What {@code restartSession()} of the LDAP domain does, minus the wait: the session is * stopped, the generation is read as the claim on it, and the lock is let go. A * {@link ReplicationDomain#restartService()} run by something else in the meantime - a * configuration change - has to leave that claim stale, or the thread which comes back * from its wait could not tell the session it stopped from the one which replaced it. */ @Test public void aClaimOnAStoppedSessionIsStaleOnceSomethingElseRestartedIt() throws Exception { final DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain domain = null; try { final int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbStaleClaim", 100); domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1); domain.disableService(); final long claim = sessionGenerationOf(domain); domain.restartService(); assertNotEquals(sessionGenerationOf(domain), claim, "a restart of the session by something else left the generation where it was," + " so the claim of the thread which stopped it still looks current"); } finally { disable(domain); remove(replServer); } } /** Read under the lock, as {@link ReplicationDomain#getSessionGeneration()} asks. */ private static long sessionGenerationOf(ReplicationDomain domain) { synchronized (domain.serviceStateLock) { return domain.getSessionGeneration(); } } /** * Publish performance test. * The test loops calling the publish methods of the ReplicationDomain. * It should not be enabled by default as it will use a lot of time. * Its call is only to investigate performance issues with the replication. */ @Test(enabled=false) public void publishPerf() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; int replServerID1 = 10; FakeReplicationDomain domain1 = null; int domain1ServerId = 1; try { int replServerPort = TestCaseUtils.findFreePort(); replServer1 = createReplicationServer(replServerID1, replServerPort, "ReplicationDomainTestDb", 100000, "localhost:" + replServerPort); SortedSet servers = newTreeSet("localhost:" + replServerPort); BlockingQueue rcvQueue1 = new LinkedBlockingQueue<>(); domain1 = new FakeReplicationDomain( testService, domain1ServerId, servers, 1000, 100000, rcvQueue1); /* * Publish a message from domain1, * Check that domain2 receives it shortly after. */ byte[] test = {1, 2, 3 ,4, 0, 1, 2, 3, 4, 5}; long timeNow = System.nanoTime(); timeNow = publishRepeatedly(domain1, test, timeNow); timeNow = publishRepeatedly(domain1, test, timeNow); timeNow = publishRepeatedly(domain1, test, timeNow); timeNow = publishRepeatedly(domain1, test, timeNow); } finally { disable(domain1); remove(replServer1); } } private long publishRepeatedly(FakeReplicationDomain domain1, byte[] test, long timeNow) { long timeStart = timeNow; for (int i = 0; i < 100000; i++) { publish(domain1, test); } timeNow = System.nanoTime(); System.out.println(timeNow - timeStart); return timeNow; } private ReplicationServer createReplicationServer(int serverId, int replicationPort, String dirName, int windowSize, String... replServers) throws Exception { return createReplicationServer(serverId, replicationPort, dirName, windowSize, newTreeSet(replServers)); } private ReplicationServer createReplicationServer(int serverId, int replicationPort, String dirName, int windowSize, SortedSet replServers) throws Exception { return new ReplicationServer( new ReplServerFakeConfiguration(replicationPort, dirName, 0, serverId, 0, windowSize, replServers)); } private void disable(ReplicationDomain... domains) throws InterruptedException { for (ReplicationDomain domain : domains) { if (domain != null) { domain.disableService(); } } } @DataProvider(name = "exportAndImportData") public Object[][] createExportAndimportData() { return new Object[][] { {1, 2}, {45610, 45720} }; } /** * Test that a ReplicationDomain is able to export and import its database * When there is only one replication server. */ @Test(dataProvider = "exportAndImportData", enabled=true) public void exportAndImport(int serverId1, int serverId2) throws Exception { final int ENTRYCOUNT=5000; DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; int replServerID = 11; FakeReplicationDomain domain1 = null; FakeReplicationDomain domain2 = null; try { int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(replServerID, replServerPort, "exportAndImportData", 100); SortedSet servers = newTreeSet("localhost:" + replServerPort); String exportedData = buildExportedData(ENTRYCOUNT); domain1 = new FakeReplicationDomain( testService, serverId1, servers, 0, exportedData, null, ENTRYCOUNT); StringBuffer importedData = new StringBuffer(); domain2 = new FakeReplicationDomain( testService, serverId2, servers, 0, null, importedData, 0); /* * Trigger a total update from domain1 to domain2. * Check that the exported data is correctly received on domain2. */ assertTrue(initializeFromRemote(domain2)); waitEndExport(exportedData, importedData); assertExportSucessful(domain1, domain2, exportedData, importedData); } finally { disable(domain1, domain2); remove(replServer); } } private boolean initializeFromRemote(ReplicationDomain domain) throws DirectoryException { try { Thread.sleep(2000); } catch (InterruptedException e) { throw new RuntimeException(e); } for (DSInfo remoteDS : domain.getReplicaInfos().values()) { if (remoteDS.getDsId() != domain.getServerId()) { domain.initializeFromRemote(remoteDS.getDsId(), NO_INIT_TASK); return true; } } return false; } /** * Test that a ReplicationDomain is able to export and import its database * across 2 replication servers. */ @Test(enabled=true) public void exportAndImportAcross2ReplServers() throws Exception { final int ENTRYCOUNT=5000; DN testService = DN.valueOf("o=test"); ReplicationServer replServer2 = null; ReplicationServer replServer1 = null; int replServerID = 11; int replServerID2 = 12; FakeReplicationDomain domain1 = null; FakeReplicationDomain domain2 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); int replServerPort1 = ports[0]; int replServerPort2 = ports[1]; replServer1 = createReplicationServer(replServerID, replServerPort1, "exportAndImportservice1", 100); replServer2 = createReplicationServer(replServerID2, replServerPort2, "exportAndImportservice2", 100, "localhost:" + replServerPort1); SortedSet servers1 = newTreeSet("localhost:" + replServerPort1); SortedSet servers2 = newTreeSet("localhost:" + replServerPort2); String exportedData = buildExportedData(ENTRYCOUNT); domain1 = new FakeReplicationDomain( testService, 1, servers1, 0, exportedData, null, ENTRYCOUNT); StringBuffer importedData = new StringBuffer(); domain2 = new FakeReplicationDomain( testService, 2, servers2, 0, null, importedData, 0); //wait for domains topology initialization { TestTimer timer = new TestTimer.Builder() .maxSleep(30, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); final FakeReplicationDomain finalDomain1 = domain1; final FakeReplicationDomain finalDomain2 = domain2; timer.repeatUntilSuccess(() -> { assertFalse(finalDomain1.getReplicaInfos().isEmpty()); assertEquals(2, finalDomain1.getRsInfos().size()); assertFalse(finalDomain2.getReplicaInfos().isEmpty()); assertEquals(2, finalDomain2.getRsInfos().size()); }); } domain2.initializeFromRemote(1, NO_INIT_TASK); waitEndExport(exportedData, importedData); assertExportSucessful(domain1, domain2, exportedData, importedData); } finally { disable(domain1, domain2); remove(replServer1, replServer2); } } @DataProvider public Object[][] replicaWithTheIdOfItsReplicationServer() { // the id of the replication server is 11 return new Object[][] { { 1, 11 }, // the importer has it: InitializeTargetMsg and the entries are sent to 11 { 11, 2 }, // the exporter has it: InitializeRequestMsg and the acknowledgements are sent to 11 }; } /** * A replica whose server id is the one of the replication server it is connected to can be * initialized, and can initialize others, through it (issue #1207). *

* The replication server takes a message for one of its replicas, not for itself: it has no data to * export or import, and before the fix it answered every total update message sent to its own id * with an ErrorMsg, so the total update never got through. */ @Test(dataProvider = "replicaWithTheIdOfItsReplicationServer") public void exportAndImportWithAReplicaHavingTheIdOfItsReplicationServer(int exporterId, int importerId) throws Exception { final int ENTRYCOUNT = 100; DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain exporter = null; FakeReplicationDomain importer = null; try { int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(11, replServerPort, "replicaWithTheIdOfItsRS", 100); SortedSet servers = newTreeSet("localhost:" + replServerPort); String exportedData = buildExportedData(ENTRYCOUNT); exporter = new FakeReplicationDomain(testService, exporterId, servers, 0, exportedData, null, ENTRYCOUNT); StringBuffer importedData = new StringBuffer(); importer = new FakeReplicationDomain(testService, importerId, servers, 0, null, importedData, 0); waitTopologyKnowsReplica(importer, exporterId); waitTopologyKnowsReplica(exporter, importerId); importer.initializeFromRemote(exporterId, NO_INIT_TASK); waitEndExport(exportedData, importedData); assertExportSucessful(exporter, importer, exportedData, importedData); } finally { disable(exporter, importer); remove(replServer); } } @DataProvider public Object[][] replicaWithTheIdOfAnotherReplicationServer() { // the replica with the id of the first replication server, 11, is the importer return new Object[][] { { 0, 1 }, // a replica of the first RS sends to 11: the first RS hands it on to the second { 1, 0 }, // a replica of the second RS sends to 11: the first RS gets it from the second RS }; } /** * A replica whose server id is the one of a replication server can be initialized across two * replication servers, whether it is connected to that replication server or to the other one * (issue #1207). */ @Test(dataProvider = "replicaWithTheIdOfAnotherReplicationServer") public void exportAndImportAcross2ReplServersWithAReplicaHavingTheIdOfAReplicationServer( int exporterRs, int importerRs) throws Exception { final int ENTRYCOUNT = 100; final int exporterId = 1; final int importerId = 11; DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; FakeReplicationDomain exporter = null; FakeReplicationDomain importer = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAnRS1", 100); replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAnRS2", 100, "localhost:" + ports[0]); String exportedData = buildExportedData(ENTRYCOUNT); exporter = new FakeReplicationDomain(testService, exporterId, newTreeSet("localhost:" + ports[exporterRs]), 0, exportedData, null, ENTRYCOUNT); StringBuffer importedData = new StringBuffer(); importer = new FakeReplicationDomain(testService, importerId, newTreeSet("localhost:" + ports[importerRs]), 0, null, importedData, 0); waitTopologyKnowsReplica(importer, exporterId); waitTopologyKnowsReplica(exporter, importerId); importer.initializeFromRemote(exporterId, NO_INIT_TASK); waitEndExport(exportedData, importedData); assertExportSucessful(exporter, importer, exportedData, importedData); } finally { disable(exporter, importer); remove(replServer1, replServer2); } } /** A replication server reports a replica which connects to it with its own server id (issue #1207). */ @Test public void replicaWithTheIdOfItsReplicationServerIsReported() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain domain = null; try { int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(11, replServerPort, "replicaWithTheIdOfItsRSIsReported", 100); TestCaseUtils.ERROR_TEXT_WRITER.clear(); domain = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + replServerPort), 0, 1); waitSharedServerIdReported(replServer, 11, testService); } finally { disable(domain); remove(replServer); } } /** * A replication server reports a replica which connects to it with the server id of a replication * server it is connected to, and only that replica (issue #1207). */ @Test public void replicaWithTheIdOfAConnectedReplicationServerIsReported() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; FakeReplicationDomain domain2 = null; FakeReplicationDomain domain11 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAConnectedRS1", 100); replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAConnectedRS2", 100, "localhost:" + ports[0]); TestCaseUtils.ERROR_TEXT_WRITER.clear(); // the domain of the second replication server connects it to the first one domain2 = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + ports[1]), 0, 1); waitTopologyKnowsReplicationServers(domain2, 2); domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1); waitSharedServerIdReported(replServer2, 11, testService); assertEquals(sharedServerIdRecords(replServer2, 2, testService), newArrayList(), "a replica whose server id no replication server has must not be reported"); } finally { disable(domain2, domain11); remove(replServer1, replServer2); } } /** * A replication server reports a replication server which connects to it with the server id of one * of its replicas (issue #1207). */ @Test public void replicationServerWithTheIdOfAConnectedReplicaIsReported() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; FakeReplicationDomain domain11 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer2 = createReplicationServer(12, ports[1], "rsWithTheIdOfAConnectedReplica2", 100); domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1); waitConnected(domain11); TestCaseUtils.ERROR_TEXT_WRITER.clear(); replServer1 = createReplicationServer(11, ports[0], "rsWithTheIdOfAConnectedReplica1", 100, "localhost:" + ports[1]); // a replication server connects to its peers for the domains it has replServer1.getReplicationServerDomain(testService, true); waitSharedServerIdReported(replServer2, 11, testService); } finally { disable(domain11); remove(replServer1, replServer2); } } /** * A replication server takes for itself the warning a peer replication server sends it when their * generation ids differ, also when one of its replicas has its server id: the replica would take the * warning for the end of the total update it is in (issue #1207). */ @Test public void generationIdWarningOfAPeerIsNotForTheReplicaWithTheIdOfTheReplicationServer() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; ReplicationBroker broker11 = null; ReplicationBroker broker2 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer1 = createReplicationServer(11, ports[0], "genIdWarningOfAPeer1", 100); replServer2 = createReplicationServer(12, ports[1], "genIdWarningOfAPeer2", 100); // a replication server takes the generation id of the first replica which connects to it broker11 = openReplicationSession(testService, 11, 100, ports[0], 1000, 1111); broker2 = openReplicationSession(testService, 2, 100, ports[1], 1000, 2222); TestCaseUtils.ERROR_TEXT_WRITER.clear(); // the second replication server connects to the first one and warns it that their generation ids differ replServer2.applyConfigurationChange(new ReplServerFakeConfiguration( ports[1], "genIdWarningOfAPeer2", 0, 12, 0, 100, newTreeSet("localhost:" + ports[0]))); waitGenerationIdWarningOfPeerReceived(testService, 11, 1111, 12, 2222); assertNoErrorMsgReceived(broker11); } finally { stop(broker11, broker2); remove(replServer1, replServer2); } } @DataProvider public Object[][] errorMsgIDsOfAPeer() { return new Object[][] { // the generation id warning, as replication servers send it since OpenDJ 3.0 and as OpenDJ 2.6 sends it { "org.opends.messages.replication-77", true }, { "14811213", true }, // another message, in either form { "org.opends.messages.replication-78", false }, { "14811214", false }, }; } /** * Of the ErrorMsgs a peer replication server sends to the server id of a replication server, only the * generation id warning stays with the replication server, in either form of the id of its message; * the replica with that server id gets the others (issue #1207). A replication server of this version * sends only the form used since OpenDJ 3.0, so the peer here sends ErrorMsgs decoded from their bytes. */ @Test(dataProvider = "errorMsgIDsOfAPeer") public void onlyTheGenerationIdWarningOfAPeerStaysWithTheReplicationServer(String msgID, boolean staysWithTheReplicationServer) throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; ReplicationBroker broker11 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer1 = createReplicationServer(11, ports[0], "errorMsgOfAPeer1", 100); replServer2 = createReplicationServer(12, ports[1], "errorMsgOfAPeer2", 100, "localhost:" + ports[0]); broker11 = openReplicationSession(testService, 11, 100, ports[0], 1000); // a replication server connects to its peers for the domains it has replServer2.getReplicationServerDomain(testService, true); waitReplicationServerConnected(replServer1, testService, 12); waitReplicationServerConnected(replServer2, testService, 11); TestCaseUtils.ERROR_TEXT_WRITER.clear(); ServerHandler toReplServer1 = replServer2.getReplicationServerDomain(testService).getConnectedRSs().get(11); String details = "error " + msgID; toReplServer1.send(errorMsg(12, 11, msgID, details)); // sent on the same connection after the first one: the replica gets the first one before it, or never String lastDetails = "last error"; toReplServer1.send(errorMsg(12, 11, "org.opends.messages.replication-78", lastDetails)); List received = receiveErrorMsgsUntil(broker11, lastDetails); if (staysWithTheReplicationServer) { assertFalse(received.contains(details), "DS(11) received the generation id warning " + msgID); waitErrorMsgReceived(details); } else { assertTrue(received.contains(details), "DS(11) did not receive the ErrorMsg " + msgID + ": " + received); } } finally { stop(broker11); remove(replServer1, replServer2); } } /** * Returns an ErrorMsg that carries the provided id of message, decoded from its bytes as a replication * server decodes it: the constructors of ErrorMsg build that id from the details. */ private static ErrorMsg errorMsg(int senderId, int destination, String msgID, String details) throws Exception { byte[] bytes = new ByteArrayBuilder() .appendByte(14) // ReplicationMsg.MSG_TYPE_ERROR .appendIntUTF8(senderId) .appendIntUTF8(destination) .appendString(msgID) .appendString(details) .appendLongUTF8(System.currentTimeMillis()) .toByteArray(); return (ErrorMsg) ReplicationMsg.generateMsg(bytes, ProtocolVersion.getCurrentVersion()); } /** Returns the details of the ErrorMsgs the replica receives before the one with the provided details. */ private List receiveErrorMsgsUntil(ReplicationBroker broker, String lastDetails) throws Exception { final List received = newArrayList(); final long deadline = System.currentTimeMillis() + SECONDS.toMillis(30); while (System.currentTimeMillis() < deadline) { final ReplicationMsg msg; try { msg = broker.receive(); } catch (SocketTimeoutException e) { continue; } assertNotNull(msg, "DS(" + broker.getServerId() + ") lost its connection"); if (msg instanceof ErrorMsg) { final String details = ((ErrorMsg) msg).getDetails().toString(); if (details.equals(lastDetails)) { return received; } received.add(details); } } throw new AssertionError("DS(" + broker.getServerId() + ") did not receive the ErrorMsg '" + lastDetails + "', only " + received); } private void waitReplicationServerConnected(ReplicationServer replServer, DN baseDN, int peerId) throws Exception { newTimer().repeatUntilSuccess(() -> assertTrue( replServer.getReplicationServerDomain(baseDN, true).getConnectedRSs().containsKey(peerId), "RS(" + peerId + ") connected to " + replServer.getMonitorInstanceName())); } /** Waits for the error log record in which a replication server reports the ErrorMsg it kept. */ private void waitErrorMsgReceived(String details) throws Exception { final String received = ERR_ERROR_MSG_RECEIVED.get(details).toString(); newTimer().repeatUntilSuccess(() -> { final List records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages(); assertTrue(records.stream().anyMatch(record -> record.contains(received)), "no replication server reported the ErrorMsg '" + details + "': " + records); }); } /** * A replication server tells its peers that a replica has left also when the replica has the server * id of one of these peers (issue #1207). */ @Test public void replicaWithTheIdOfAPeerReplicationServerIsForgottenWhenItLeaves() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer1 = null; ReplicationServer replServer2 = null; FakeReplicationDomain domain1 = null; FakeReplicationDomain domain11 = null; try { int[] ports = TestCaseUtils.findFreePorts(2); replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAPeerLeaves1", 100); replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAPeerLeaves2", 100, "localhost:" + ports[0]); domain1 = new FakeReplicationDomain(testService, 1, newTreeSet("localhost:" + ports[0]), 0, 1); domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1); waitTopologyKnowsReplica(domain1, 11); domain11.disableService(); waitTopologyForgetsReplica(domain1, 11); } finally { disable(domain1, domain11); remove(replServer1, replServer2); } } private static TestTimer newTimer() { return new TestTimer.Builder() .maxSleep(30, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); } private void waitConnected(ReplicationDomain domain) throws Exception { newTimer().repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "DS(" + domain.getServerId() + ") connected")); } private void waitTopologyForgetsReplica(ReplicationDomain domain, int dsId) throws Exception { newTimer().repeatUntilSuccess(() -> assertFalse(domain.getReplicaInfos().containsKey(dsId), "DS(" + dsId + ") has left, and DS(" + domain.getServerId() + ") still knows it")); } private void waitTopologyKnowsReplicationServers(ReplicationDomain domain, int count) throws Exception { newTimer().repeatUntilSuccess(() -> assertEquals(domain.getRsInfos().size(), count, "replication servers known to DS(" + domain.getServerId() + ")")); } private void waitSharedServerIdReported(ReplicationServer replServer, int serverId, DN baseDN) throws Exception { newTimer().repeatUntilSuccess(() -> assertFalse(sharedServerIdRecords(replServer, serverId, baseDN).isEmpty(), replServer.getMonitorInstanceName() + " did not report that DS(" + serverId + ") has the id of an RS")); } /** * Waits for the error log record in which a replication server reports the warning a peer sent it * because their generation ids differ. */ private void waitGenerationIdWarningOfPeerReceived(DN baseDN, int rsId, long rsGenerationId, int peerId, long peerGenerationId) throws Exception { // the warning names the address the peer sees the replication server at, which is not known here final String[] warning = WARN_BAD_GENERATION_ID_FROM_RS.get( rsId, "@", rsGenerationId, baseDN, peerId, peerGenerationId).toString().split("@"); final String received = ERR_ERROR_MSG_RECEIVED.get(warning[0]).toString(); newTimer().repeatUntilSuccess(() -> { for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) { if (record.contains(received) && record.contains(warning[1])) { return; } } fail("RS(" + rsId + ") did not receive the warning of RS(" + peerId + "): " + TestCaseUtils.ERROR_TEXT_WRITER.getMessages()); }); } private void assertNoErrorMsgReceived(ReplicationBroker broker) { try { ReplicationMsg msg; while ((msg = broker.receive()) != null) { assertFalse(msg instanceof ErrorMsg, "DS(" + broker.getServerId() + ") received " + msg); } } catch (SocketTimeoutException expected) { // nothing more to receive } } /** Returns the records of the error log in which the replication server reports the provided replica. */ private List sharedServerIdRecords(ReplicationServer replServer, int serverId, DN baseDN) { final String message = WARN_REPLICA_HAS_THE_SERVER_ID_OF_A_REPLICATION_SERVER.get( replServer.getMonitorInstanceName(), serverId, baseDN, serverId).toString(); final List records = newArrayList(); for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) { if (record.contains(message)) { records.add(record); } } return records; } /** * 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); } } /** InitializeTask double recording the completion reported by the domain. */ private static final class RecordingInitializeTask extends InitializeTask { private final CountDownLatch completed = new CountDownLatch(1); private volatile DirectoryException failure; @Override public void updateTaskCompletionState(DirectoryException de) { failure = de; completed.countDown(); } DirectoryException waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException { assertTrue(completed.await(timeout, unit), "the initialize task never completed"); return failure; } } /** * An ErrorMsg answering an initialization request can be created within the * same millisecond as the requester's import/export context when the whole * topology runs on one host: it must terminate the pending initialization * instead of being discarded as stale (issue #861). */ @Test(enabled=true) public void errorMsgFromSameMillisecondTerminatesPendingInitialize() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain domain2 = null; ReplicationBroker broker3 = null; try { int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(13, replServerPort, "sameMillisecondErrorMsgDb", 100); SortedSet servers = newTreeSet("localhost:" + replServerPort); domain2 = new FakeReplicationDomain( testService, 2, servers, 0, null, new StringBuffer(), 0); broker3 = openReplicationSession(testService, 3, 100, replServerPort, 10000, domain2.getGenerationID()); waitTopologyKnowsReplica(domain2, 3); RecordingInitializeTask task = new RecordingInitializeTask(); domain2.initializeFromRemote(3, task); long startTime = domain2.getImportExportContext().getStartTime(); // strictly older than the context: still ignored as stale ErrorMsg staleError = new ErrorMsg(3, 2, LocalizableMessage.raw("stale error")); staleError.setCreationTime(startTime - 1); broker3.publish(staleError); // same millisecond as the context: must terminate the initialization ErrorMsg currentError = new ErrorMsg(3, 2, LocalizableMessage.raw("current error")); currentError.setCreationTime(startTime); broker3.publish(currentError); DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS); assertNotNull(failure, "the initialization completed without an error"); assertEquals(failure.getMessageObject().toString(), "current error", "the ErrorMsg timestamped before the context must stay ignored"); assertFalse(domain2.ieRunning(), "the terminated initialization must release the import/export context"); } finally { stop(broker3); disable(domain2); remove(replServer); } } /** * When the initialization request receives no answer at all - the publish * was silently dropped or the answer was lost - the stalled-request watchdog * must fail the task after the configured delay instead of letting it wait * forever (issue #861). */ @Test(enabled=true) public void stalledInitializeFromRemoteIsAborted() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; FakeReplicationDomain domain2 = null; ReplicationBroker broker3 = null; try { int replServerPort = TestCaseUtils.findFreePort(); replServer = createReplicationServer(14, replServerPort, "stalledInitializeRequestDb", 100); SortedSet servers = newTreeSet("localhost:" + replServerPort); domain2 = new FakeReplicationDomain( testService, 2, servers, 0, null, new StringBuffer(), 0); // broker3 receives the InitializeRequestMsg and never answers it broker3 = openReplicationSession(testService, 3, 100, replServerPort, 10000, domain2.getGenerationID()); waitTopologyKnowsReplica(domain2, 3); RecordingInitializeTask task = new RecordingInitializeTask(); domain2.initializeFromRemote(3, task); assertFalse(domain2.abortStalledInitializeFromRemote(60000), "the initialization must not be aborted before the delay elapses"); final FakeReplicationDomain requester = domain2; TestTimer abortTimer = new TestTimer.Builder() .maxSleep(30, SECONDS) .sleepTimes(10, MILLISECONDS) .toTimer(); abortTimer.repeatUntilSuccess(() -> assertTrue( requester.abortStalledInitializeFromRemote(50), "the stalled initialization was never aborted")); DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS); assertNotNull(failure, "the stalled initialization must fail the task"); assertEquals(failure.getMessageObject().toString(), ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(testService, 3).toString()); assertFalse(domain2.ieRunning(), "the aborted initialization must release the import/export context"); assertFalse(domain2.abortStalledInitializeFromRemote(0), "a second abort must be a no-op once the context is released"); } finally { stop(broker3); disable(domain2); remove(replServer); } } private void waitTopologyKnowsReplica(ReplicationDomain domain, int dsId) throws Exception { TestTimer timer = new TestTimer.Builder() .maxSleep(30, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); timer.repeatUntilSuccess(() -> assertTrue(domain.getReplicaInfos().containsKey(dsId), "DS(" + dsId + ") is not known to the domain")); } private String buildExportedData(final int ENTRYCOUNT) { final StringBuilder sb = new StringBuilder(); for (int i = 0; i < ENTRYCOUNT; i++) { sb.append("key : value").append(i).append("\n\n"); } return sb.toString(); } private void waitEndExport(String exportedData, StringBuffer importedData) throws Exception { int count = 0; while (importedData.length() < exportedData.length() && count < 500*5) { if(count % 100 == 0) { //capture thread dump on start and every 10 seconds logger.debug(LocalizableMessage.raw("waitEndExport: thread dump on count=" + count)); logger.debug(LocalizableMessage.raw(generateThreadDump())); } count ++; Thread.sleep(100); } } private void assertExportSucessful(FakeReplicationDomain domain1, FakeReplicationDomain domain2, String exportedData, StringBuffer importedData) { assertEquals(getLeftEntryCount(domain2), 0, "Wrong LeftEntryCount for export"); assertEquals(getLeftEntryCount(domain1), 0, "Wrong LeftEntryCount for import"); assertEquals(importedData.length(), exportedData.length()); assertEquals(importedData.toString(), exportedData); /* * The direction of the total update, as each side reported it while it was in it. * ieRunning() cannot tell the two apart, and something which is guarding the data of a * replica has to: the exporter keeps its data and its ServerState - and keeps replaying * into them - while the importer is having both replaced. */ assertEquals(domain2.importReportedDuringImport(), Boolean.TRUE, "the replica whose data is being replaced must report a total update into itself"); assertEquals(domain1.importReportedDuringExport(), Boolean.FALSE, "the replica which is exporting must not report a total update into itself:" + " its data and its ServerState are left alone"); } private long getLeftEntryCount(ReplicationDomain domain) { final ImportExportContext ieContext = domain.getImportExportContext(); if (ieContext != null) { return ieContext.getLeftEntryCount(); } return 0; // import/export is finished } /** * Sender side of the Total Update Perf test. * The goal of this test is to measure the performance * of the total update code. * It is not intended to be run as part of the daily unit test but * should only be used manually by developer in need of testing the * performance improvement or non-regression of the total update code. * Use this test in combination with the receiverInitialize() : * - enable the test * - start the senderInitialize first using * ./build.sh \ * -Dorg.opends.test.suppressOutput=false \ * -Dtest.methods=org.opends.server.replication.service.ReplicationDomainTest.senderInitialize test * - start the receiverInitialize second. * - you may want to change HOST1 and HOST2 to use 2 different hosts * if you don't want to do a loopback test. * - don't forget to disable again the tests after running them */ final String HOST1 = "localhost:"; final String HOST2 = "localhost:"; final int SENDERPORT = 10102; final int RECEIVERPORT = 10101; @Test(enabled=false) public void senderInitialize() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; int replServerID = 12; FakeStressReplicationDomain domain1 = null; try { SortedSet servers = newTreeSet(HOST1 + SENDERPORT, HOST2 + RECEIVERPORT); replServer = createReplicationServer(replServerID, SENDERPORT, "ReplicationDomainTestDb", 100, servers); BlockingQueue rcvQueue1 = new LinkedBlockingQueue<>(); domain1 = new FakeStressReplicationDomain( testService, 2, servers, 1000, rcvQueue1); System.out.println("waiting"); Thread.sleep(1000000000); } finally { disable(domain1); remove(replServer); } } /** * See comments in senderInitialize() above. */ @Test(enabled=false) public void receiverInitialize() throws Exception { DN testService = DN.valueOf("o=test"); ReplicationServer replServer = null; int replServerID = 11; FakeStressReplicationDomain domain1 = null; try { SortedSet servers = newTreeSet(HOST1 + SENDERPORT, HOST2 + RECEIVERPORT); replServer = createReplicationServer(replServerID, RECEIVERPORT, "ReplicationDomainTestDb", 100, servers); BlockingQueue rcvQueue1 = new LinkedBlockingQueue<>(); domain1 = new FakeStressReplicationDomain( testService, 1, servers, 100000, rcvQueue1); /* * Trigger a total update from domain1 to domain2. * Check that the exported data is correctly received on domain2. */ while (!initializeFromRemote(domain1)) { System.out.println("trying..."); Thread.sleep(1000); } System.out.println("waiting"); Thread.sleep(10000000); } finally { disable(domain1); remove(replServer); } } }