* It has to be long enough for a domain which does not wait for the replay to have saved * its ServerState by the time the change is applied - that is the failure the test * reports - and well under the time a domain which does wait gives the replay, which that * test leaves at its default. Spent inside that wait, so what it costs the test is itself * and nothing more. *
* Counted from the moment the session was cut rather than from the wait, because that is * the only moment this test can see: {@code ReplicationBroker.stop()} is the first * statement of {@code disableService()}, and it stops the domain being connected before * the listener thread is asked to stop and joined - a join with no bound on it - and * before the ServerState is saved. So what this delay has to outlast is that whole * remainder of {@code disable()} and not the save alone. The remainder is a millisecond * on an idle machine and hundreds of them on a loaded one, and a delay of the same order * would hand the released replay a race against the save rather than a loss to it: the * change would be recorded whether or not anything waited for it, and the test would stop * saying anything without ever failing. */ private static final long SETTLE_BEFORE_RELEASE_IN_MS = 2000; /** * How long a domain is told to wait for a replay it can not drain, in the test which * checks that it gives up rather than hold the task which is taking it down. Long enough * to be told apart from not waiting at all, short enough for a test to spend. *
* Told apart from the rest of {@code disable()}, to be exact: the test times that call as
* a whole rather than the wait inside it, so this budget is what has to dominate cutting
* the session, joining the listener thread with no bound on the join, and saving the
* ServerState with an internal modify. That remainder is a millisecond on an idle machine
* and hundreds of them on a loaded one, and a budget of the same order would have the
* assertion satisfied by the remainder alone - the wait taken out of the domain and
* nothing reporting it.
*/
private static final long TEST_REPLAY_DRAIN_TIMEOUT_IN_MS = 2000;
/** An entry with a entryUUID. */
private Entry personWithUUIDEntry;
private Entry personWithSecondUniqueID;
private Entry user3Entry;
private DN user3dn;
private String user3UUID;
private String baseUUID;
private DN user1dn;
private String user1entrysecondUUID;
private String user1entryUUID;
/** A "person" entry. */
private Entry personEntry;
private int replServerPort;
private String domain1uid;
private String domain2uid;
private String domain3uid;
private DN domain1dn;
private DN domain2dn;
private DN domain3dn;
private Entry domain1;
private Entry domain2;
private Entry domain3;
private int domainSid = 55;
private DN baseDN;
/** Set up the environment for performing the tests in this Class. */
@BeforeClass
@Override
public void setUp() throws Exception
{
super.setUp();
baseDN = DN.valueOf("ou=People," + TEST_ROOT_DN_STRING);
// Create necessary backend top level entry
TestCaseUtils.addEntry(
"dn: " + baseDN,
"objectClass: top",
"objectClass: organizationalUnit",
"entryUUID: 11111111-1111-1111-1111-111111111111");
baseUUID = getEntryUUID(baseDN);
replServerPort = TestCaseUtils.findFreePort();
// replication server
String replServerLdif =
"dn: cn=Replication Server, " + SYNCHRO_PLUGIN_DN + "\n"
+ "objectClass: top\n"
+ "objectClass: ds-cfg-replication-server\n"
+ "cn: Replication Server\n"
+ "ds-cfg-replication-port: " + replServerPort + "\n"
+ "ds-cfg-replication-db-directory: UpdateOperationTest\n"
+ "ds-cfg-replication-server-id: 107\n"
/*
* Long enough for a delivery which a test stops on its way through the replay:
* the acks of an assured update are waited for from the moment it is published,
* and the default second is spent long before a test which parks that delivery
* has let go of it. Nothing waits it out - no test here leaves an assured update
* unacknowledged - so it only bounds a failure.
*/
+ "ds-cfg-assured-timeout: 120000ms\n";
// suffix synchronized
String testName = "updateOperationTest";
String synchroServerLdif =
"dn: cn=" + testName + ", cn=domains, " + SYNCHRO_PLUGIN_DN + "\n"
+ "objectClass: top\n"
+ "objectClass: ds-cfg-replication-domain\n"
+ "cn: " + testName + "\n"
+ "ds-cfg-base-dn: " + baseDN + "\n"
+ "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
+ "ds-cfg-server-id: "+ domainSid +"\n"
+ "ds-cfg-receive-status: true\n";
configureReplication(replServerLdif, synchroServerLdif);
}
private void testSetUp(String tc) throws Exception
{
personEntry = TestCaseUtils.makeEntry(
"dn: uid=user.1." + tc + "," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.1",
"homePhone: 951-245-7634",
"description: This is the description for Aaccf Amar.",
"st: NC",
"mobile: 027-085-0537",
"postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762",
"mail: user.1@example.com",
"cn: Aaccf Amar",
"l: Rockford",
"pager: 508-763-4246",
"street: 17984 Thirteenth Street",
"telephoneNumber: 216-564-6748",
"employeeNumber: 1",
"sn: Amar",
"givenName: Aaccf",
"postalCode: 85762",
"userPassword: password",
"initials: AA");
/*
* The 2 entries defined in the following code are used for the naming
* conflict resolution test (called namingConflicts)
* They must have the same DN but different entryUUID.
*/
user1entryUUID = "33333333-3333-3333-3333-333333333333";
user1entrysecondUUID = "22222222-2222-2222-2222-222222222222";
user1dn = DN.valueOf("uid=user1" + tc + "," + baseDN);
personWithUUIDEntry = TestCaseUtils.makeEntry(
"dn: " + user1dn,
"objectClass: top", "objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson", "uid: user.1",
"homePhone: 951-245-7634",
"description: This is the description for Aaccf Amar.", "st: NC",
"mobile: 027-085-0537",
"postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.1@example.com",
"cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246",
"street: 17984 Thirteenth Street",
"telephoneNumber: 216-564-6748", "employeeNumber: 1",
"sn: Amar", "givenName: Aaccf", "postalCode: 85762",
"userPassword: password", "initials: AA",
"entryUUID: " + user1entryUUID + "\n");
personWithSecondUniqueID = TestCaseUtils.makeEntry(
"dn: "+ user1dn,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.1",
"homePhone: 951-245-7634",
"description: This is the description for Aaccf Amar.",
"st: NC",
"mobile: 027-085-0537",
"postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762",
"mail: user.1@example.com",
"cn: Aaccf Amar",
"l: Rockford",
"pager: 508-763-4246",
"street: 17984 Thirteenth Street",
"telephoneNumber: 216-564-6748",
"employeeNumber: 1",
"sn: Amar",
"givenName: Aaccf",
"postalCode: 85762",
"userPassword: password",
"initials: AA",
"entryUUID: "+ user1entrysecondUUID);
user3UUID = "44444444-4444-4444-4444-444444444444";
user3dn = DN.valueOf("uid=user3" + tc + "," + baseDN);
user3Entry = TestCaseUtils.makeEntry("dn: "+ user3dn,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.1",
"homePhone: 951-245-7634",
"description: This is the description for Aaccf Amar.",
"st: NC",
"mobile: 027-085-0537",
"postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762",
"mail: user.3@example.com",
"cn: Aaccf Amar",
"l: Rockford",
"pager: 508-763-4246",
"street: 17984 Thirteenth Street",
"telephoneNumber: 216-564-6748",
"employeeNumber: 1",
"sn: Amar",
"givenName: Aaccf",
"postalCode: 85762",
"userPassword: password",
"initials: AA",
"entryUUID: " + user3UUID);
domain1dn = DN.valueOf("dc=domain1," + baseDN);
domain2dn = DN.valueOf("dc=domain2,dc=domain1," + baseDN);
domain3dn = DN.valueOf("dc=domain3,dc=domain1," + baseDN);
domain1 = TestCaseUtils.makeEntry(
"dn:" + domain1dn,
"objectClass:domain",
"dc:domain1");
domain2 = TestCaseUtils.makeEntry(
"dn:" + domain2dn,
"objectClass:domain",
"dc:domain2");
domain3 = TestCaseUtils.makeEntry(
"dn:" + domain3dn,
"objectClass:domain",
"dc:domain3");
}
/** Add an entry in the database. */
private CSN addEntry(Entry entry) throws Exception
{
AddOperation addOp = connection.processAdd(entry);
assertEquals(addOp.getResultCode(), ResultCode.SUCCESS);
assertNotNull(getEntry(entry.getName(), 1000, true));
return OperationContext.getCSN(addOp);
}
/** Delete an entry in the database. */
private void delEntry(DN dn) throws Exception
{
connection.processDelete(dn);
assertNull(getEntry(dn, 1000, false));
}
/**
* Tests whether the synchronization provider receive status can be disabled
* then re-enabled.
* FIXME Enable this test when broker suspend/resume receive are implemented.
*/
@Test(enabled=false)
public void toggleReceiveStatus() throws Exception
{
testSetUp("toggleReceiveStatus");
logger.error(LocalizableMessage.raw("Starting synchronization test : toggleReceiveStatus"));
/*
* Open a session to the replicationServer using the broker API.
* This must use a different serverId to that of the directory server.
*/
final int serverId = 2;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
// Disable the directory server receive status.
setReceiveStatus(synchroServerEntry.getName(), false);
// Create and publish an update message to add an entry.
broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID));
assertNull(getEntry(personWithUUIDEntry.getName(), 1000, true),
"The replication message was replayed while it should not have been: "
+ "the server receive status was disabled");
// Enable the directory server receive status.
setReceiveStatus(synchroServerEntry.getName(), true);
broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID));
assertNotNull(getEntry(personWithUUIDEntry.getName(), 10000, true),
"The replication message was not replayed while it should have been: "
+ "the server receive status was reenabled");
// Delete the entries to clean the database.
broker.publish(
new DeleteMsg(personWithUUIDEntry.getName(), gen.newCSN(), user1entryUUID));
assertNull(getEntry(personWithUUIDEntry.getName(), 10000, false),
"The DELETE replication message was not replayed");
}
finally
{
broker.stop();
}
}
private AddMsg addMsg(CSNGenerator gen, Entry entry, String uniqueId, String parentId)
{
return new AddMsg(gen.newCSN(), entry.getName(), uniqueId, parentId,
entry.getObjectClassAttribute(), entry.getAllAttributes(), null);
}
/**
* Tests whether the synchronization provider fails over when it loses
* the heartbeat from the replication server.
*/
@Test
public void lostHeartbeatFailover() throws Exception
{
testSetUp("lostHeartbeatFailover");
logger.error(LocalizableMessage.raw("Starting replication test : lostHeartbeatFailover"));
/*
* Open a session to the replicationServer using the broker API.
* This must use a different serverId to that of the directory server.
*/
int serverId = 2;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
// Create and publish an update message to add an entry.
broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID));
assertNotNull(getEntry(personWithUUIDEntry.getName(), 30000, true),
"The ADD replication message was not replayed");
// Send a first modify operation message.
List
* The change is parked inside the operation it is replayed by, so the thread is caught
* while it still owns it rather than raced for: a change which is released by the
* ordinary recovery instead ends the same way - delivered again and applied - so a test
* which only watched the end state would pass whether or not the hand-back happened.
*
* What tells them apart is which thread replays the change next. The thread which held
* it is gone, and the change is replayed by one of the threads which replaced it, so
* the delivery it is replayed from can only be a new one: an attempt which the same
* thread made again would be the retry in place, and a change nobody handed back is
* never delivered again at all - it stays listed as owned by a thread which is gone,
* with the ServerState of this domain stopped behind it for good.
*/
@Test
public void aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain() throws Exception
{
testSetUp("aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain"));
final int serverId = 14;
/*
* In the group of the replication server, so that the delete published below is one
* this domain has to acknowledge: an assured update from a broker of another group is
* acknowledged by the replication server itself, and says nothing about the replay.
*/
ReplicationBroker broker =
openAssuredReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.909," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.909",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
domain.resetUnreplayedChangeAlertThrottle();
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
/*
* Only the counters of the replay are read: a session restart takes this domain
* through NOT_CONNECTED, which resets every monitoring counter of the replication
* service - the updates it received and processed, and the assured acks it sent -
* and handing a change back is a session restart.
*/
final long initialApplied = getMonitorAttrValue(baseDN, "replayed-updates-ok");
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
/*
* Hold the replayed deletes where they are. The park is taken at the pre-parse
* plugin point, which runs inside op.run() and before anything of the backend was
* taken: the replay thread stops there while it still owns the change, and the
* pre-operation plugins are not invoked for synchronization operations at all.
*/
final CSN csn = gen.newCSN();
final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
OperationType.DELETE, "PreParse", op -> csn.equals(OperationContext.getCSN(op)));
final AtomicReference
* It cannot run on the thread of the test: stopping the replay threads joins them, so it
* does not return while one of them is parked in a change, which is the whole point of
* this fixture.
*
* @param replayThreads how many replay threads the server must run, or {@code null} for
* as many as it computes on its own
* @param failure where the reconfiguration reports what it ran into, if anything
* @return the thread which is doing the reconfiguration
*/
private static Thread startReplayThreadReconfiguration(
final Integer replayThreads, final AtomicReference
* The messages which come first are discarded: this broker is told about everything the
* replication server has for it, and the ack of one change is what is being looked for.
*
* @param broker the broker the change was published on
* @param csn the change the ack is expected for
* @return the ack of that change
* @throws Exception if it never arrived
*/
private static AckMsg awaitAck(final ReplicationBroker broker, final CSN csn) throws Exception
{
final long deadline = System.nanoTime() + SECONDS.toNanos(60);
while (deadline - System.nanoTime() > 0)
{
final ReplicationMsg msg;
try
{
msg = broker.receive();
}
catch (SocketTimeoutException e)
{
// The broker reads under a timeout of its own, which is far shorter than the
// budget here: a quiet second is not an answer.
continue;
}
if (msg == null)
{
// The broker stopped rather than timed out: there is nothing left to read from,
// and reading it again would spin a core for the rest of the budget.
throw new AssertionError("the session " + csn + " was published on is gone,"
+ " so the ack of that change can no longer arrive");
}
if (msg instanceof AckMsg && csn.equals(((AckMsg) msg).getCSN()))
{
return (AckMsg) msg;
}
}
throw new AssertionError("the delivery of " + csn + " was never acknowledged");
}
/**
* Waits for the delivery which took over from the abandoned one to be parked, reporting
* what the reconfiguration ran into when that is why nothing was parked.
*
* A reconfiguration which throws once it has stopped the replay threads leaves this
* server with no replay thread at all: nothing can be parked then, and the timeout of
* the wait would be reported in place of the failure which brought it about.
*
* @param parked the park the delivery is expected to be caught in
* @param failure where the reconfiguration reports what it ran into, if anything
* @return the thread which is replaying the parked operation
* @throws Exception if no operation was parked in time
*/
private static Thread awaitParkedOrReportReconfigurationFailure(
final ParkedReplay parked, final AtomicReference
* Waiting for that, rather than for the thread to be started, is what puts the change in
* the hands of a thread which has already been asked to stop: every replay thread is
* asked to stop before the first of them is joined, so a thread which is joining has
* asked the parked one. Where the thread waits is checked as well as that it waits: the
* state on its own would be satisfied by any wait at all, including one taken before the
* replay threads were asked to stop.
*
* @param reconfiguration the thread which is changing the number of replay threads
* @param failure where that thread reports what it ran into, if anything
* @throws Exception if it never reached the join
*/
private static void awaitStoppingTheReplayThreads(
final Thread reconfiguration, final AtomicReference
* A backend which is failing fails every change in flight, which is what this test
* reproduces with two changes. A count kept for the last failed change only is reset
* by each of them in turn, so the give up would never be reached and this replica
* would restart its session to the replication server without end.
*/
@Test
public void everyChangeWhichCanNotBeReplayedIsGivenUpOn() throws Exception
{
testSetUp("everyChangeWhichCanNotBeReplayedIsGivenUpOn");
logger.error(LocalizableMessage.raw("Starting replication test : everyChangeWhichCanNotBeReplayedIsGivenUpOn"));
final int serverId = 13;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry first = TestCaseUtils.addEntry(
"dn: uid=user.889.1," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.889.1",
"cn: Aaccf Amar",
"sn: Amar");
Entry second = TestCaseUtils.addEntry(
"dn: uid=user.889.2," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.889.2",
"cn: Aaccf Amar",
"sn: Amar");
String firstUuid = getEntry(first.getName(), 1, true).parseAttribute("entryuuid").asString();
String secondUuid = getEntry(second.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
// Two changes have to be given up on here, so the budget is shortened the same way.
setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
try
{
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
final CSN firstCSN = gen.newCSN();
final CSN secondCSN = gen.newCSN();
broker.publish(new DeleteMsg(first.getName(), firstCSN, firstUuid));
broker.publish(new DeleteMsg(second.getName(), secondCSN, secondUuid));
TestTimer giveUpTimer = new TestTimer.Builder()
.maxSleep(120, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
giveUpTimer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(firstCSN),
"the replica did not give up on the first change it can never replay");
assertTrue(domain.getServerState().cover(secondCSN),
"the replica did not give up on the second change it can never replay");
}
});
assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 2,
"both changes must be counted as failed, once each");
/*
* Two changes counted more than once each climb past +2, and the poll which lands
* on it would pass: the value has to be seen to stay put.
*/
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 2,
"both changes must be counted as failed, once each");
assertNotNull(getEntry(first.getName(), 1, true), "the first entry must not have been deleted");
assertNotNull(getEntry(second.getName(), 1, true), "the second entry must not have been deleted");
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
resetReplayGiveUpDelay();
}
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 901]: a replica whose replay give-up budget is unlimited keeps
* asking for a change it can not replay instead of ever recording it as replayed, and
* the budget is read from the configuration for every decision - so lowering it takes
* effect on the change which is failing right now, without this server being restarted.
*/
@Test
public void anUnlimitedReplayGiveUpDelayIsNeverSpent() throws Exception
{
testSetUp("anUnlimitedReplayGiveUpDelayIsNeverSpent");
logger.error(LocalizableMessage.raw("Starting replication test : anUnlimitedReplayGiveUpDelayIsNeverSpent"));
final int serverId = 18;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.901," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.901",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
// An operator who would rather have this domain stop than have it diverge: the
// change is retried for as long as it keeps failing.
setReplayGiveUpDelay("unlimited");
try
{
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
final CSN csn = gen.newCSN();
broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
/*
* One delivery burns IN_PLACE_REPLAY_ATTEMPTS short circuits before it is handed
* back and the session is restarted, so twice that many of them is a change which
* was delivered, given back and delivered again - the loop this replica is
* deliberately left in.
*/
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(100, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")
> 2 * IN_PLACE_REPLAY_ATTEMPTS,
"the change was not asked for again while the budget was unlimited");
}
});
assertFalse(domain.getServerState().cover(csn),
"a replica which never gives up must not record a change it did not apply");
/*
* The monitor attribute is the only signal this domain has left: a budget which
* is never spent raises no alert and counts no failed replay, so the change this
* replica keeps asking for is visible here and nowhere else. It counts changes
* rather than deliveries, so the redeliveries above leave it at one.
*/
assertMonitorAttrValueEventually(baseDN, "changes-with-failed-replay", 1,
"the change which keeps failing must be counted, once, as a failing change");
/*
* The change was delivered at least twice by now, so a counter which was bumped
* per delivery rather than per change would already be past the value which is
* being watched: what this asserts is carried by the value itself rather than by
* how long the window is, which is why the default number of samples is enough
* here - a window covering a redelivery would have to outlast a backoff which has
* been climbing since the first failure.
*/
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
"no change may be counted as failed while the budget is unlimited");
/*
* The budget is read for every decision, so the failure which comes next spends
* this one: the change this replica was holding on to is given up on without the
* server, or the domain, being restarted for the new value to be seen.
*/
setReplayGiveUpDelay("0ms");
TestTimer giveUpTimer = new TestTimer.Builder()
.maxSleep(120, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
giveUpTimer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(csn),
"a budget which was lowered must be spent by the change which is failing");
}
});
assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
"the change which was given up on must be counted once");
assertMonitorAttrValueEventually(baseDN, "changes-with-failed-replay", 0,
"a change which was given up on leaves the pending changes and stops being counted");
assertNotNull(getEntry(tmp.getName(), 1, true), "the entry must not have been deleted");
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
resetReplayGiveUpDelay();
}
}
finally
{
broker.stop();
}
}
/**
* The result codes a replay is retried on rather than skipped: the storage failing to
* serve the operation, and a lock which could not be taken (OPENDJ-885) - the ten
* in-place attempts only yield to the thread holding it, so a lock held for a while
* burns every one of them and the change is as absent from the data as after a storage
* failure.
*/
@DataProvider(name = "transientReplayFailures")
public Object[][] transientReplayFailures()
{
return new Object[][] {
{ ResultCode.UNAVAILABLE, 14, "user.889.3" },
{ ResultCode.BUSY, 15, "user.889.4" },
};
}
/**
* Test case for [Issue 889]: a replay which fails on the server itself has the session
* restarted and the change delivered again, and a failure which clears in the meantime
* has the change applied exactly once, without the change being given up on and without
* it being reported as failed.
*/
@Test(dataProvider = "transientReplayFailures")
public void transientReplayFailureIsRetriedAndTheChangeApplied(
final ResultCode transientFailure, final int serverId, final String uid) throws Exception
{
testSetUp("transientReplayFailureIsRetriedAndTheChangeApplied." + uid);
logger.error(LocalizableMessage.raw(
"Starting replication test : transientReplayFailureIsRetriedAndTheChangeApplied "
+ transientFailure));
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=" + uid + "," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: " + uid,
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
final long initialReplayed = getMonitorAttrValue(baseDN, "replayed-updates-ok");
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
/*
* The backend is unavailable the way it is while a rebuild is performed or while it
* is offline (OPENDJ-49), and it stays unavailable for longer than the replay is
* retried in place: the change is only applied if the session is restarted and the
* replication server delivers it a second time.
*/
try
{
// Registered inside the try which deregisters it: the plugin is consulted for
// every delete in this server, so one left behind fails the tests which follow.
ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
transientFailure.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
final CSN csn = gen.newCSN();
broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
assertNull(getEntry(tmp.getName(), 30000, false),
"the change was not replayed once the backend served the operation again");
Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
.as("the change must have been delivered again after the session was restarted")
.isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
TestTimer timer = new TestTimer.Builder()
.maxSleep(30, SECONDS)
.sleepTimes(100, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(csn),
"a change which was replayed must be recorded as replayed");
}
});
assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1,
"the change must be recorded as replayed");
/*
* A change applied twice - the delivery which failed and the one which took over
* from it, the OPENDJ-1115 regression the takeover is there to prevent - takes the
* counter through +1 on its way to +2, so the value has to be seen to stay put
* rather than to be reached once. It has to be watched for longer than the
* session restart which brings that second delivery, too, or the assertion stops
* looking before the delivery it is looking for could arrive.
*/
assertMonitorAttrValueStays(baseDN, "replayed-updates-ok", initialReplayed + 1,
MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
"a change which was delivered again must be applied exactly once");
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
"a change which was replayed after a transient failure must not count as failed");
assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
"a transient failure must not tell the administrator that this replica diverged");
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
}
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 889]: the result code the server puts on an internal error is
* configurable and is not validated as a result code, so it can be set to one conflict
* resolution knows how to solve. Such a change is left to conflict resolution, and when
* that can not solve it either the change is retried as the storage failure it is -
* recording it as replayed after one attempt would be issue #889 again.
*/
@Test
public void changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried() throws Exception
{
testSetUp("changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried");
logger.error(LocalizableMessage.raw(
"Starting replication test : changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried"));
final int serverId = 16;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.889.5," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.889.5",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
domain.resetUnreplayedChangeAlertThrottle();
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
/*
* UNWILLING_TO_PERFORM is one of the codes solveNamingConflict(ModifyDNOperation)
* solves, so it must not be treated as a failure of the server before conflict
* resolution had its chance - and it is what the storage reports here.
*/
// Put back whatever was configured, not the default: a suite which runs with
// another server-error-result-code must not be rewritten by this test.
final int previousServerErrorResultCode =
getServerContext().getCoreConfigManager().getServerErrorResultCode().intValue();
try
{
/*
* Changed inside the try which puts it back: the result code this server reports
* an internal error with is server-wide, so one left behind would change which
* road every later replay of this suite takes.
*/
setServerErrorResultCode(ResultCode.UNWILLING_TO_PERFORM.intValue());
/*
* The failure lasts longer than the attempts made in place, so the change is only
* applied if it was left out of the ServerState and delivered again rather than
* recorded as replayed once conflict resolution reported it could not be solved.
*/
ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
ResultCode.UNWILLING_TO_PERFORM.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
final CSN csn = gen.newCSN();
broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
assertNull(getEntry(tmp.getName(), 120000, false),
"the change was skipped rather than retried once the storage served the operation");
Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
.as("the change must have been delivered again rather than recorded as replayed")
.isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
"a change which was replayed in the end must not be counted as given up on");
assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
"a change which was replayed in the end must not tell the administrator that this replica diverged");
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
setServerErrorResultCode(previousServerErrorResultCode);
}
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 889]: a change whose message can not be turned into an operation
* must not hold this replica's ServerState back for good.
*
* There is no operation to retry and no delivery which would decode any better, so the
* change has to be skipped rather than left listed as the barrier: a change which stays
* uncommitted holds back the ServerState - and every change which follows it, from
* every master - and the delivery which would replace it is turned down while a replay
* thread still owns it, so nothing would ever move it again.
*/
@Test
public void aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack() throws Exception
{
testSetUp("aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack"));
final int serverId = 17;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.889.6," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.889.6",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
domain.resetUnreplayedChangeAlertThrottle();
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
final CSN csn = gen.newCSN();
broker.publish(undecodableModifyMsg(csn, tmp.getName(), uuid));
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(csn),
"a change which can never be decoded must not hold the ServerState back");
}
});
assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
"a change which could not be decoded must be counted as failed");
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 1,
"a change which could not be decoded must be counted once");
Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
.as("the administrator must be told that this replica now diverges")
.isGreaterThan(initialAlerts);
}
finally
{
broker.stop();
}
}
/**
* Builds a ModifyMsg which travels the protocol intact and can not be turned into an
* operation.
*
* The encoded modifications are carried as an opaque byte array and are only read by
* {@code createOperation()}, so a message whose modifications are corrupt is decoded,
* listed as pending and handed to a replay thread before it fails - which is the point
* of this test.
*
* @param csn the CSN to give the change
* @param dn the entry the change is on
* @param entryUUID the UUID of that entry
* @return a message whose replay can not build an operation
* @throws Exception if the message could not be built
*/
private ModifyMsg undecodableModifyMsg(CSN csn, DN dn, String entryUUID) throws Exception
{
final List
* Which of the two roads a failure takes is decided by the operation rather than by
* its CSN: a message no operation could be built from will not build one on the next
* delivery either, so it is given up on where it is reported, while an operation which
* was built may well have reached the backend - so its change is kept out of the
* ServerState and asked for again, wherever in the replay the failure happened. The
* entry DN of a ModifyMsg which does not parse is that case: it leaves
* {@code getEntryDN()} null and the replay throws before the CSN of the operation is
* read, so a give-up keyed off that CSN would record a change which never reached the
* backend as replayed, which is this issue by another route.
*/
@Test
public void aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed() throws Exception
{
testSetUp("aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed"));
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.889.7," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.889.7",
"cn: Aaccf Amar",
"sn: Amar");
final DN dn = tmp.getName();
final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
domain.resetUnreplayedChangeAlertThrottle();
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
final CSNGenerator gen = new CSNGenerator(18, TimeThread.getTime());
final CSN csn = gen.newCSN();
final String description = "the replay must fail once the operation is built";
final List
* The change reaches the backend, so a ServerState which excludes it records nowhere
* that it was applied: the replication server sends it again when the domain is enabled
* back, and a change which is already in the data is replayed a second time - resolved
* as a conflict, or left as a conflict entry when the changes around it were resent with
* it and their dependency ordering was forgotten along with the pending changes.
*/
@Test
public void aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled() throws Exception
{
testSetUp("aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeBeingAppliedIsRecordedBeforeTheDomainIsDisabled"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
final CSNGenerator gen = new CSNGenerator(serverId, 0);
final Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.908," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.908",
"cn: Aaccf Amar",
"sn: Amar");
final DN dn = tmp.getName();
final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final CSN csn = gen.newCSN();
boolean disableAttempted = false;
try
{
/*
* Park the replay inside op.run(): the pre-parse plugin point is reached once the
* replay thread started applying the change and before the change reaches the
* backend, which is the window this issue is about. The pre-operation point would
* not do - it is not invoked for synchronization operations.
*/
PausePreParsePlugin.pause(OperationType.DELETE, dn);
broker.publish(new DeleteMsg(dn, csn, uuid));
assertTrue(PausePreParsePlugin.awaitPaused(OperationType.DELETE, 60, SECONDS),
"the replay thread never started applying the change");
assertTrue(domain.isConnected(),
"this test needs a domain which is still up when the change is being applied");
/*
* Let the parked replay finish once the domain is inside the wait for it, so that
* the change reaches the backend while the ServerState is about to be saved. The
* session is cut after the flag is set and immediately before that wait, and well
* before the state is saved, so a domain which is not connected anymore is one which
* is about to wait for this very change.
*
* Released a moment after that rather than on the disconnection itself, and this is
* what makes the test decide rather than guess: a domain which does not wait - the
* lock taken out of the replay, or the state saved before the wait as it was before
* this fix - has saved its ServerState long before the delay is out, so the change
* lands after that save and the assertion below reports it. Releasing on the
* disconnection instead handed the replay the join of the listener thread as a head
* start, which is enough for it to be recorded by a domain which never waited.
*
* The delay is spent inside the wait, so it costs this test nothing and holds
* whatever budget it needs to be well under REPLAY_DRAIN_TIMEOUT_IN_MS.
*/
final Thread releaser =
releaseWhenDisconnected(domain, OperationType.DELETE, SETTLE_BEFORE_RELEASE_IN_MS);
disableAttempted = true;
try
{
domain.disable();
}
finally
{
releaser.join(SECONDS.toMillis(60));
}
/*
* The entry is gone, so the change did reach the backend: getEntry() waits for it
* and reports it, since a domain which did not wait for the replay lets it finish
* a moment later rather than not at all.
*/
getEntry(dn, 30000, false);
/*
* Read the ServerState which was saved rather than the one in memory: disable()
* clears the in-memory one, and the saved one is what the domain reads back when
* it is enabled again - and what the replication server resumes this replica from.
* Read it before the domain is enabled back, or the change being sent again and
* replayed a second time would make the state cover it either way, which is the
* very outcome this test is about.
*/
assertTrue(persistedServerState().cover(csn),
"a change which reached the backend must be recorded in the saved ServerState");
}
finally
{
PausePreParsePlugin.release(OperationType.DELETE);
if (disableAttempted)
{
/*
* Only when disable() was reached, and whether or not it got to the end: setting
* the flag is its first act, so a disable() which threw half way through still
* left a domain which has to be enabled back. Enabling one which was never
* disabled is what must not happen - it would reload the ServerState and start a
* broker which is already running, behind the back of the tests which follow.
*/
domain.enable();
}
}
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 908]: a domain which can not get the replay of its changes to
* finish goes down anyway rather than holding the administrative task which is taking it
* down - an import, a restore, a backend being taken offline - for as long as a backend
* which stopped answering takes to answer.
*
* The change may then reach the backend without being recorded in the ServerState, which
* is what the warning in the log says: the replication server sends it again once the
* domain is enabled back, which this test also checks, since a domain which gave up on
* the wait must still end up consistent.
*/
@Test
public void theDomainStopsWaitingForAReplayWhichDoesNotFinish() throws Exception
{
testSetUp("theDomainStopsWaitingForAReplayWhichDoesNotFinish");
logger.error(LocalizableMessage.raw(
"Starting replication test : theDomainStopsWaitingForAReplayWhichDoesNotFinish"));
final int serverId = 20;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
final CSNGenerator gen = new CSNGenerator(serverId, 0);
final Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.908.2," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.908.2",
"cn: Aaccf Amar",
"sn: Amar");
final DN dn = tmp.getName();
final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final CSN csn = gen.newCSN();
final long drainTimeout = domain.getReplayDrainTimeout();
boolean disableAttempted = false;
try
{
/*
* A replay which does not finish is waited out for as long as an operation can be
* waiting for the entry it is on - the best part of twenty seconds: this test can
* not, so the domain gives up on the wait after a moment instead. Set inside the
* try which puts it back, like the pause below: both are the domain's and the
* server's for as long as they are left behind.
*/
domain.setReplayDrainTimeout(TEST_REPLAY_DRAIN_TIMEOUT_IN_MS);
PausePreParsePlugin.pause(OperationType.DELETE, dn);
broker.publish(new DeleteMsg(dn, csn, uuid));
assertTrue(PausePreParsePlugin.awaitPaused(OperationType.DELETE, 60, SECONDS),
"the replay thread never started applying the change");
// The replay is parked and stays parked: the domain has to come down all the same.
final long startedAt = System.nanoTime();
disableAttempted = true;
domain.disable();
final long waitedMs = NANOSECONDS.toMillis(System.nanoTime() - startedAt);
/*
* Only this test releases the pause, and it has not done so yet, so an operation
* still parked here is one the domain came down without waiting for - which is what
* the give-up is. Read before the finally below releases it.
*/
Assertions.assertThat(PausePreParsePlugin.parkedCount(OperationType.DELETE))
.as("the domain must have come down while the replay was still being applied")
.isEqualTo(1);
/*
* Measured against the default this test overrode rather than against a copy of
* its value: an override which stopped taking effect would have the domain wait
* the whole default out, and that is what this has to catch.
*/
assertTrue(waitedMs < drainTimeout,
"the domain waited " + waitedMs + " ms for a replay it can not drain,"
+ " which is not short of the " + drainTimeout + " ms it waits by default");
/*
* And the wait was taken rather than skipped: the replay is parked for good, so a
* domain which really waits for it spends the whole budget it was given. Without
* this the test reports the same thing whether the domain waited for the changes in
* flight or never waited for anything - the give-up is only half of what a bounded
* wait is.
*/
assertTrue(waitedMs >= TEST_REPLAY_DRAIN_TIMEOUT_IN_MS,
"the domain came down in " + waitedMs + " ms, so it did not wait the "
+ TEST_REPLAY_DRAIN_TIMEOUT_IN_MS + " ms it was given for the replay of a"
+ " change which was still being applied");
}
finally
{
PausePreParsePlugin.release(OperationType.DELETE);
domain.setReplayDrainTimeout(drainTimeout);
if (disableAttempted)
{
domain.enable();
}
}
/*
* The entry goes away: the replay the domain gave up on was released by the finally
* above and finished after the ServerState had been saved, which is what the give-up
* costs. This says the change is in the data - not that it was delivered again, since
* the delete which does it is the first replay rather than the second.
*/
getEntry(dn, 30000, false);
/*
* The change is in the data and in no ServerState, so the replication server owns it
* still and sends it again over the session which the domain being enabled back
* brought up. Replaying it a second time is the cost of the wait running out, and
* conflict resolution absorbs it - what must not happen is the replica staying behind
* for good. The state coming to cover the CSN is what evidences that delivery: the
* domain forgot the change with its pending changes, so nothing else records it.
*/
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(csn),
"the change must be recorded once it has been delivered again");
}
});
}
finally
{
broker.stop();
}
}
/**
* Starts a thread which releases the operations parked by
* {@link PausePreParsePlugin} once the domain has cut its session - which it does on its
* way down, immediately before it waits for the replay of the changes in flight - plus a
* delay which puts the release inside that wait rather than ahead of it.
*
* @param domain the domain which is about to be taken down
* @param operation the type of operation the pause was registered for
* @param settleInMs how long to wait after the session was cut before the parked
* operations are released, which has to be well under the time the
* domain waits for them and longer than the rest of {@code disable()} -
* the session being cut is its first act, so the listener thread being
* joined and the ServerState being saved are both inside this delay
* @return the thread, already started
*/
private Thread releaseWhenDisconnected(
final LDAPReplicationDomain domain, final OperationType operation, final long settleInMs)
{
final Thread releaser = new Thread(new Runnable()
{
@Override
public void run()
{
/*
* Bounded, and a daemon: a domain which never goes down - because taking it down
* threw - must not leave a thread spinning for the rest of the run. The pause has
* a bound of its own, so the parked operation is released either way.
*/
final long deadline = System.nanoTime() + SECONDS.toNanos(60);
try
{
while (domain.isConnected() && System.nanoTime() - deadline < 0)
{
Thread.sleep(1);
}
/*
* The session is cut, so the domain is on its way to the wait for the replay:
* give it that long to get there and, if it is not waiting for anything, to save
* the ServerState this change must be in.
*/
Thread.sleep(settleInMs);
PausePreParsePlugin.release(operation);
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
}, "issue 908 replay releaser");
releaser.setDaemon(true);
releaser.start();
return releaser;
}
/**
* Returns the ServerState of the test domain as it is saved in the backend.
*
* That is the one the domain reads back when it is enabled again, and the one the
* replication server resumes this replica from - the in-memory one is cleared by
* {@code disable()}.
*
* @return the ServerState read from the base entry of the domain
* @throws Exception if the base entry could not be read
*/
private ServerState persistedServerState() throws Exception
{
final ServerState persisted = new ServerState();
for (String value : getEntry(baseDN, 1, true).parseAttribute("ds-sync-state").asSetOfString())
{
persisted.update(new CSN(value));
}
return persisted;
}
/**
* A ModifyMsg whose operation can not tell which change it carries.
*
* The operation is built - so the replay is past the point where a message is given up
* on - and its entry DN does not parse, which is what has
* {@code ModifyOperationBasis.getEntryDN()} return null and the replay throw before
* {@code OperationContext.getCSN(op)} is reached. Such a message can not travel the
* protocol: the DN of a ModifyMsg is decoded on the way in and the operation is built
* from its {@code toString()}, so this one is handed to the domain rather than
* published.
*/
private static final class ModifyMsgWithAnUnparseableOperationDN extends ModifyMsg
{
private ModifyMsgWithAnUnparseableOperationDN(
CSN csn, DN dn, List
* A backend under maintenance is waited out for minutes, which no test can afford: the
* budget is spent over a couple of deliveries instead. The domain reads the property for
* every decision it makes, so this takes effect on a change which is failing right now,
* and none of the values it takes stops or starts the session.
*
* The domain outlives the test methods, so a test which shortens the budget puts it back
* with {@link #resetReplayGiveUpDelay()} in a finally, and calls this one before that
* try: the reset then only ever runs on an attribute which is there to be removed.
*
* @param delay
* the budget in the duration syntax of the property: {@code 2000ms},
* {@code 0ms} to give up as soon as one delivery of the change failed,
* {@code unlimited} never to give up on it
*/
private void setReplayGiveUpDelay(String delay)
{
modifyDomainConfig(synchroServerEntry.getName(), REPLACE, ATTR_REPLAY_GIVE_UP_DELAY, delay);
}
/**
* Puts the configured replay give-up budget of the domain of this test back, that is the
* default of the property: the domain outlives the test methods, so a shortened budget
* which is left behind is the next test's too.
*/
private void resetReplayGiveUpDelay()
{
modifyDomainConfig(synchroServerEntry.getName(), DELETE, ATTR_REPLAY_GIVE_UP_DELAY);
}
/**
* Enable or disable the receive status of a synchronization provider.
*
* @param syncConfigDN The DN of the synchronization provider configuration
* entry.
* @param enable Specifies whether the receive status should be enabled
* or disabled.
*/
private static void setReceiveStatus(DN syncConfigDN, boolean enable)
{
modifyDomainConfig(syncConfigDN, REPLACE, "ds-cfg-receive-status", enable ? "TRUE" : "FALSE");
}
/**
* Test that the ReplicationDomain (plugin inside LDAP server) adjust
* its internal CSN generator to the last CSN received. Steps:
* - create a domain with the current date in the CSN generator
* - make it receive an update with a CSN in the future
* - do a local operation replicated on that domain
* - check that the update generated for that operation has a CSN in the future.
*/
@Test(enabled=true)
public void csnGeneratorAdjust() throws Exception
{
testSetUp("csnGeneratorAdjust");
logger.error(LocalizableMessage.raw("Starting synchronization test : CSNGeneratorAdjust"));
/*
* Open a session to the replicationServer using the broker API.
* This must use a different serverId to that of the directory server.
*/
final int serverId = 88;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
consumeAllMessages(broker); // clean leftover messages from lostHeartbeatFailover()
try
{
final long inTheFuture = System.currentTimeMillis() + (3600 * 1000);
CSNGenerator gen = new CSNGenerator(serverId, inTheFuture);
// Create and publish an update message to add an entry.
AddMsg addMsg = addMsg(gen, user3Entry, user3UUID, baseUUID);
broker.publish(addMsg);
// Check that the entry has not been created in the directory server.
assertNotNull(getEntry(user3Entry.getName(), 1000, true),
"The entry has not been created");
// Modify the entry
connection.processModify(modifyRequest(user3Entry.getName(), REPLACE, "telephonenumber", "01 02 45"));
// See if the client has received the msg
ReplicationMsg msg = broker.receive();
Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
ModifyMsg modMsg = (ModifyMsg) msg;
assertTrue(modMsg.getCSN().getTimeSec()-addMsg.getCSN().getTimeSec()<=1,
"The MOD timestamp should have been adjusted to the ADD one");
// Delete the entries to clean the database.
broker.publish(
new DeleteMsg(user3Entry.getName(), gen.newCSN(), user3UUID));
// Check that the delete operation has been applied.
assertNull(getEntry(user3Entry.getName(), 10000, false),
"The DELETE replication message was not replayed");
}
finally
{
broker.stop();
}
}
/**
* Consumes all the messages sent to this broker. This is useful at the start
* of a test to avoid leftover messages from previous test runs.
*/
private void consumeAllMessages(ReplicationBroker broker)
{
final List