/*
|
* 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<String> servers = newTreeSet("localhost:" + replServerPort1);
|
BlockingQueue<UpdateMsg> rcvQueue1 = new LinkedBlockingQueue<>();
|
domain1 = new FakeReplicationDomain(
|
testService, domain1ServerId, servers, 100, 1000, rcvQueue1);
|
|
SortedSet<String> servers2 = newTreeSet("localhost:" + replServerPort2);
|
BlockingQueue<UpdateMsg> 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<Integer, ServerState> states1 = domain1.getReplicaStates();
|
ServerState state2 = states1.get(domain2ServerId);
|
assertNotNull(state2, "getReplicaStates is not showing DS2");
|
|
Map<Integer, ServerState> 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<Integer, DSInfo> 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<String> servers = newTreeSet("localhost:" + replServerPort);
|
BlockingQueue<UpdateMsg> 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<String> 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<String> 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<String> servers1 = newTreeSet("localhost:" + replServerPort1);
|
SortedSet<String> 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).
|
* <p>
|
* 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<String> 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<String> 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<String> receiveErrorMsgsUntil(ReplicationBroker broker, String lastDetails) throws Exception
|
{
|
final List<String> 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<String> 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<String> 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<String> 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<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);
|
}
|
}
|
|
/** 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<String> 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<String> 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<String> servers =
|
newTreeSet(HOST1 + SENDERPORT, HOST2 + RECEIVERPORT);
|
|
replServer = createReplicationServer(replServerID, SENDERPORT,
|
"ReplicationDomainTestDb", 100, servers);
|
|
BlockingQueue<UpdateMsg> 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<String> servers =
|
newTreeSet(HOST1 + SENDERPORT, HOST2 + RECEIVERPORT);
|
|
replServer = createReplicationServer(replServerID, RECEIVERPORT,
|
"ReplicationDomainTestDb", 100, servers);
|
|
BlockingQueue<UpdateMsg> 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);
|
}
|
}
|
|
}
|
