* 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; /** * How long after a session restart threw on the state checkpointer the test which reads * the level the request was given back with looks at the session. *
* The checkpointer takes the request back on its next tick, a second later, and holds
* the backoff it was given back with - a second, the first time - before it starts the
* session: two seconds after the throw, then. A request given back without the backoff
* has the session up a second after the throw instead. Halfway between the two, so that
* either outcome has half a second to be told apart in. A machine slow enough to push the
* tick past this delay has the test look before the session could be back under either
* level, which proves less and reports nothing false.
*/
private static final long INTO_THE_BACKOFF_IN_MS = 1500;
/** 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 925]: a session restart which could not run is run again, so
* that the change it was asked for is delivered again rather than left waiting for a
* delivery which can not come.
*
* The request is taken by the thread which runs the restart before the restart runs -
* a change released while a restart is under way is not one that restart asks for - so
* a restart which throws where it starts the session again used to take the request
* away with it. Nothing asked for it a second time: the session had been stopped and
* was not started back, and the domain stayed out of the topology, with the change
* still owned by the replication server and the ServerState of this replica stopped
* behind it, until the server was restarted.
*
* Two restarts fail rather than one, so that both threads which run one meet a failure:
* the first is spent by the replay thread which released the change, and the request it
* gives back is run by the state checkpointer, whose restart is the second to fail. The
* checkpointer reports that one and runs the request again a moment later - which is
* what delivers the change - so a checkpointer which ended on the failure instead would
* leave the change where the replay thread left it.
*/
@Test
public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception
{
testSetUp("aSessionRestartWhichCouldNotRunIsRunAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aSessionRestartWhichCouldNotRunIsRunAgain"));
final int serverId = 24;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.925," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.925",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long initialReplayed = getMonitorAttrValue(baseDN, "replayed-updates-ok");
try
{
/*
* The backend is unavailable for longer than the replay is retried in place, so
* the change is only applied if the session is restarted and the replication
* server delivers it again - and the first two restarts the domain runs for it
* fail the way a broken enableService() does, leaving the session stopped: the one
* the replay thread runs, and the one the state checkpointer runs for it.
*/
ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
domain.failNextSessionRestarts(2);
final CSN csn = gen.newCSN();
final List
* The restart which fails is the checkpointer's, so the two levels are told apart by
* when the session is back: on the checkpointer's next tick, a second after the throw,
* when the request was given back as it was made, and one backoff later when it was
* given back with the wait it is owed. Halfway between the two the session is still down
* under the second and up under the first.
*
* That the restart which fails is the checkpointer's is read off the error log rather
* than assumed: only the checkpointer reports a restart which threw on it as
* {@code ERR_REPLAY_SESSION_RESTART_FAILED}, so one such report says the thread on its
* way out asked for the restart rather than ran it, on every run and not by the clock -
* a thread which ran it itself would spend the failure at once, and the checkpointer's
* own restart would then be one which runs.
*/
@Test
public void aRestartAskedForWithoutTheBackoffIsGivenBackWithIt() throws Exception
{
testSetUp("aRestartAskedForWithoutTheBackoffIsGivenBackWithIt");
logger.error(LocalizableMessage.raw(
"Starting replication test : aRestartAskedForWithoutTheBackoffIsGivenBackWithIt"));
final int serverId = 25;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.925.backoff," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.925.backoff",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
/*
* The replayed delete is held where it is, so that the replay thread is stopped
* while it owns the change: see aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain
* for the fixture.
*/
final CSN csn = gen.newCSN();
final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
OperationType.DELETE, "PreParse", op -> csn.equals(OperationContext.getCSN(op)));
final AtomicReference
* 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. An
* operation which can not be given the ManageDsaIT control the replay adds to every one
* of them is that case: it 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"));
assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseOperationRefusesAControl::new,
18, "user.889.7", "the replay must fail once the operation is built");
}
/**
* The errors a replay meets which say nothing about the changes which follow: a class
* which could not be linked, and a stack which ran out on the entry being replayed. The
* second is an error of the JVM, but the thread which met it is whole again once the
* stack has unwound and the entry is what raised it, so ending the thread would have one
* change this replica can not replay cost it one replay thread per delivery.
*
* Each row builds its error where it is thrown rather than here, so that the stack trace
* it carries is the one of the replay it unwound.
*/
@DataProvider(name = "recoverableReplayErrors")
public Object[][] recoverableReplayErrors()
{
return new Object[][] {
{ (Supplier
* A change is owned by the replay thread which took it, and that ownership is what
* keeps a change being replayed from being replayed a second time (OPENDJ-1115): every
* later delivery of it is refused as a duplicate. Ownership is given back on the roads
* which run to their end, so an Error - which unwinds the replay out of every one of
* them - would leave the change listed, uncommitted and owned by a thread which is not
* replaying it anymore: nobody could replay it, and this domain's ServerState would
* never move past it again.
*
* The error is thrown where the operation is built, so what these rows pin is the arm
* which reports it and takes the road of a failed replay: it is caught inside the replay
* and never leaves it. The roads which do leave it - the give-back of a replay which was
* unwound, and the widened catch of the replay thread which is what keeps that thread
* alive - are pinned by
* {@link #aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain()}, where the throw is
* made past the point any catch of the replay runs on.
*/
@Test(dataProvider = "recoverableReplayErrors")
public void aChangeWhoseReplayThrewAnErrorIsDeliveredAgain(
final Supplier
* An OutOfMemoryError is not turned into a failed replay and reported the way the other
* errors are - building the report asks for more of what the JVM has run out of - and
* the thread it unwinds is not replaced. The change it was replaying must not go with
* it: it is handed back so that the delivery which follows can replay it.
*/
@Test
public void aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain() throws Exception
{
testSetUp("aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain"));
/*
* The threads are read by identity rather than counted: what this test is about is the
* thread which met the error being gone, which is what #923 sanctions. Whether the pool
* is refilled afterwards is the half of that issue which is left open, and a count
* would freeze it here as the behaviour which is wanted.
*/
final Set
* At least one rather than exactly one: the alert is raised for every thread of the
* server which ends that way, and a thread of some other component ending during the
* test must not turn this into a failure of the wrong test.
*/
private static void assertUncaughtExceptionAlertRaisedSince(int initialAlerts, String message)
{
Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION))
.as(message)
.isGreaterThanOrEqualTo(initialAlerts + 1);
}
/**
* Test case for [Issue 922]: a change whose ack could not be published still takes the
* road its own replay decided.
*
* The ack of a delivery is published in a finally which every road out of the replay runs
* through, and it is published on the session that delivery came over - which is being
* torn down when the replay of the change failed. Whether the change was applied is not
* something that publish can tell, so a throw there is reported and the replay carries on
* to the road the change itself decided: this one failed, so it is kept out of the
* ServerState, counted and asked for again.
*
* Left to unwind, that throw would step over the give-back which follows it and leave the
* change owned by a thread which is not replaying it anymore.
*/
@Test
public void aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain() throws Exception
{
testSetUp("aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain"));
assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseAckThrows::new,
21, "user.922.3", "the ack of this change throws on the way out of the replay");
}
/**
* Test case for [Issue 922] and [Issue 923]: an OutOfMemoryError met where the ack of a
* delivery is published ends the replay thread, the way one met by the replay itself does.
*
* A throw from the ack is caught so that it does not unwind the replay past the give-back
* of the change and past the hand-out of the changes which were waiting for it. A JVM
* which has run out of memory is the one exception to that: it is not something to carry
* on replaying from, so it is left to end this thread - the change is given back on the
* way out, and the uncaught exception handler of DirectoryThread writes the line and
* raises the alert #923 is about. Caught like every other throw from there, it would have
* this thread replay the changes which follow on an exhausted heap, with nothing said
* anywhere.
*/
@Test
public void aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain() throws Exception
{
testSetUp("aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain"));
final Set
* The catches of the replay itself span the roads which decide what became of the change,
* and the ack is published once that is decided. What the replay runs afterwards - the
* give-back of a change which failed, and the hand-out of the changes which were waiting
* for it - is past every one of them: a throw there unwinds {@code replay()} with the
* change still owned by this thread, and a change owned by a thread which is not replaying
* it anymore is refused as a duplicate on every later delivery. So it is given back on the
* way out, and the error is left to the replay thread, whose catch is what keeps it alive
* for the changes which follow (issue #923).
*/
@Test
public void aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain() throws Exception
{
testSetUp("aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain"));
final Set
* The change is given back on the road a failed replay takes rather than handed back bare,
* so the failure counts against the give-up budget of the change: a change which keeps
* unwinding the replays it is given to is eventually recorded as one this replica could
* not apply, and the administrator is told that it now diverges. Handed back bare it would
* be asked for, and have this domain restart its session for it, for as long as the server
* is up - which is the wedge this issue is about wearing another face.
*/
@Test
public void aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn() throws Exception
{
testSetUp("aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn"));
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.922.7," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.922.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);
setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
try
{
final CSN csn = new CSNGenerator(25, TimeThread.getTime()).newCSN();
final List
* A change which is waiting for another one is handed out by {@code getNextUpdate()},
* which the replay runs once it is done with the change it was given - after the ack of
* that delivery has been published. A throw from that ack used to unwind the replay past
* it, and the change had been committed by then, so nothing was owed back and nothing was
* handed out: the changes parked behind it stayed parked, and the ServerState of this
* domain stayed behind them until some other change was replayed here.
*
* What tells the two apart is that the parked change is replayed at all. The child is
* seen parked before the parent is let go, so the replay queue is not where it can come
* from anymore: only {@code getNextUpdate()} hands it out, and with the parent held until
* then, the thread which committed the parent is, as a rule, the one left to call it - a
* change nobody handed out is replayed by no one at all, and the wait below is what says so.
* Which thread replays it is deliberately not asserted: {@code getNextUpdate()} hands a
* parked change to whichever thread calls it first once the changes before it are gone,
* and the thread which parked it calls it on its own way out, so a parker which is slow
* to get there takes the child back itself when the parent commits in between.
*
* The pin this gives issue #922 holds on that first arm alone: nothing here orders the
* parker's own {@code getNextUpdate()} call against the parent being let go, and a parker
* delayed past the parent's release, commit and ack takes the child back on its own way
* out instead - measured with a mutant, 500 ms after {@code checkDependencies()} parks the
* child. A revert of the fix this test is for goes uncaught on that arm: the wait above
* sees the child parked either way, and only the parent's road runs the code #922 is about.
*/
@Test
public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
{
testSetUp("theChangesParkedBehindAChangeWhoseAckFailedAreReplayed");
logger.error(LocalizableMessage.raw(
"Starting replication test : theChangesParkedBehindAChangeWhoseAckFailedAreReplayed"));
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
final String parentUUID = "26262626-2626-2626-2626-262626262626";
final String childUUID = "27272727-2727-2727-2727-272727272727";
final Entry parent = TestCaseUtils.makeEntry(
"dn: ou=parked.922," + baseDN,
"objectClass: top",
"objectClass: organizationalUnit",
"ou: parked.922",
"entryUUID: " + parentUUID);
final Entry child = TestCaseUtils.makeEntry(
"dn: uid=user.922.8,ou=parked.922," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.922.8",
"cn: Aaccf Amar",
"sn: Amar",
"entryUUID: " + childUUID);
/*
* Both adds are held at the pre-parse plugin point, one at a time. The first park is
* what keeps the parent listed as pending - and owned by the thread replaying it - while
* the child is checked for dependencies, so the child is parked behind a change which is
* in flight rather than behind one which has already been applied.
*/
final CSN parentCsn = gen.newCSN();
final CSN childCsn = gen.newCSN();
final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
OperationType.ADD, "PreParse",
op -> parentCsn.equals(OperationContext.getCSN(op))
|| childCsn.equals(OperationContext.getCSN(op)));
final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
try
{
domain.processUpdate(new AddMsgWhoseAckThrows(parentCsn, parent.getName(), parentUUID,
baseUUID, parent.getObjectClassAttribute(), parent.getAllAttributes()));
parked.awaitParked(60, SECONDS);
domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
child.getObjectClassAttribute(), child.getAllAttributes(), null));
/*
* Seen parked before the parent is let go. Released on the spot, the parent could be
* applied and committed before a replay thread has taken the child off the queue at
* all, and the child would then be replayed from the queue with nothing to wait for -
* a pass which says nothing about the hand-out this test is about.
*/
assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", initialDependent + 1,
"the child must be parked behind the parent while the parent is held");
/*
* The parent is applied and its ack throws where it is published. The replay carries
* on all the same, and the child is the change it hands out next.
*/
parked.release();
parked.awaitParked(60, SECONDS);
parked.release();
assertNotNull(getEntry(child.getName(), 30000, true),
"the change which was parked behind the one whose ack threw must be applied");
}
finally
{
parked.deregister();
}
}
/**
* Test case for [Issue 922]: a change handed out as a dependency is given back when the
* replay it was handed to is unwound.
*
* A change which was parked behind another one is handed out by {@code getNextUpdate()}
* to whichever thread calls it first once the changes before it are gone, and that
* thread owns it from then on. The give-back on the way out of an unwound replay asks
* which change this thread owns, so the hand-out has to be recorded where that question
* is answered, not only on the change: left out, the change would stay owned by a thread
* which is not replaying it anymore, and every later delivery of it would be refused as
* a duplicate - the wedge of this issue, on the dependency road.
*
* The parent is held at the pre-parse plugin point while the child is delivered, and it
* is let go only once the child is seen parked behind it: that is what says the child was
* handed out rather than taken off the queue, since a parked change leaves by
* {@code getNextUpdate()} and by no other road. The thread which is thrown out of the
* child is read at the same plugin point, for the assertion that the Error did not end
* it; which thread it is says nothing about the hand-out and is not asserted - the parker
* calls {@code getNextUpdate()} on its own way out, so it takes the child back itself
* when the parent commits before it gets there. The child is thrown out of once, inside
* the replay, and then unwound past its ack, on the road every catch of the replay has
* already run on.
*/
@Test
public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() throws Exception
{
testSetUp("aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound"));
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final CSNGenerator gen = new CSNGenerator(29, TimeThread.getTime());
final String parentUUID = "29292929-2929-2929-2929-292929292929";
final String childUUID = "30303030-3030-3030-3030-303030303030";
final Entry parent = TestCaseUtils.makeEntry(
"dn: ou=handed-out.922," + baseDN,
"objectClass: top",
"objectClass: organizationalUnit",
"ou: handed-out.922",
"entryUUID: " + parentUUID);
final Entry child = TestCaseUtils.makeEntry(
"dn: uid=user.922.11,ou=handed-out.922," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.922.11",
"cn: Aaccf Amar",
"sn: Amar",
"entryUUID: " + childUUID);
final CSN parentCsn = gen.newCSN();
final CSN childCsn = gen.newCSN();
final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
OperationType.ADD, "PreParse", op -> parentCsn.equals(OperationContext.getCSN(op)));
/*
* The thread which reaches the plugin point with the child is the one replaying it, and
* it is read there rather than parked: a park would hold the child where the throw is
* made, and it is the throw which is wanted.
*/
final AtomicReference
* An assured write in SAFE_READ mode is told that its change is durable here by the ack
* this replica publishes, and it is published in a finally which every road out of the
* replay runs through - the roads an Error unwinds among them. A replica which is asking
* for a change again must never have told a master that the change is in its data, so the
* ack of a delivery whose replay threw reports the error and names this replica.
*
* The error is thrown from a plugin point which runs inside the operation, so the change
* travels a real session and the ack can be read off the broker which published it - the
* counters can not report it, since handing the change back restarts the session and that
* resets every one of them.
*/
@Test
public void theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt() throws Exception
{
testSetUp("theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt");
logger.error(LocalizableMessage.raw(
"Starting replication test : theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt"));
final int serverId = 28;
/*
* In the group of the replication server, so that what is 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.922.9," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.922.9",
"cn: Aaccf Amar",
"sn: Amar");
final String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final CSN csn = gen.newCSN();
/*
* Thrown out of one replay and no more: the delivery which takes over from the one
* which was unwound is what applies the change, and a change this replica could never
* replay would be given up on rather than acknowledged twice.
*/
final ThrownFromReplay thrown = ShortCircuitPlugin.throwFromReplayedOperations(
OperationType.DELETE, "PreParse",
op -> csn.equals(OperationContext.getCSN(op)),
() -> new LinkageError("the replay of this change meets an Error"), 1);
try
{
final DeleteMsg delete = new DeleteMsg(tmp.getName(), csn, uuid);
delete.setAssured(true);
delete.setAssuredMode(AssuredMode.SAFE_READ_MODE);
broker.publish(delete);
final AckMsg ack = awaitAck(broker, csn);
assertTrue(ack.hasReplayError(),
"the ack of a delivery whose replay threw an Error must report it rather than be"
+ " the plain ack a master would take for a durable write");
assertFalse(ack.hasTimeout(),
"the ack must be the one the delivery published, not the one the replication"
+ " server makes up when it gives up waiting for it");
Assertions.assertThat(ack.getFailedServers())
.as("the replica whose replay threw must be the one the ack names")
.containsExactly(domainSid);
assertEquals(thrown.thrownCount(), 1, "the replay of the change must have been unwound");
}
finally
{
thrown.deregister();
}
/*
* The change was given back rather than recorded as replayed, so the delivery which
* follows applies it - which is what the ack above said had not happened yet.
*/
assertNull(getEntry(tmp.getName(), 30000, false),
"the change must be applied by the delivery which took over from the one whose"
+ " replay threw");
assertTrue(domain.getServerState().cover(csn),
"the change must be recorded as replayed once it has been applied");
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 922]: a change whose replay keeps throwing an Error is given up
* on rather than asked for forever.
*
* An Error takes the road every other failed replay takes, so the failures it leaves
* behind count against the give-up budget of the change: a change this replica can never
* apply must not hold its ServerState - and every change which follows it, from every
* master - back for good, whether its replay reported the failure or threw it.
*/
@Test
public void aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn() throws Exception
{
testSetUp("aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn"));
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.922.4," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.922.4",
"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);
setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
try
{
final CSN csn = new CSNGenerator(22, TimeThread.getTime()).newCSN();
final List
* A change which waits for another one is parked and stays owned by the replay thread
* which parked it, while that thread goes on to the changes which follow: it is handed
* out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
* it is done, so it is replayed by whichever thread calls it first once the change it was
* waiting for is gone - the thread which cleared it, as a rule.
* A replay which is unwound leaves the thread which parked it without that road - it
* takes the next delivery off the shared queue instead, and never comes back to the
* change it parked - and every redelivery of a change a replay thread owns is refused as
* a duplicate. On a domain which then goes quiet that change is where this replica's
* ServerState, and every change behind it from every master, stops.
*
* The replay which is unwound here is one which applied its change: the change it was
* replaying is committed and owns nothing anymore by the time the give-back on the way
* out of {@code replay()} runs, so the restart that give-back asks for is the one thing
* which has the parked change delivered again - a replay which failed would have asked
* for the same restart on the road of its own change. The parked change travels the
* replication server, and nothing but a new session brings it back.
*/
@Test
public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception
{
testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain"));
final DN waitedOn = addEntryForChange("user.954.1");
final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
final DN other = addEntryForChange("user.954.2");
final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
domain.resetUnreplayedChangeAlertThrottle();
final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
"no change of this domain is waiting for another one when this test starts");
// The replica the changes come from: one of its own, so that the CSNs of this test are
// never covered by the ServerState another one left behind.
final int serverId = 30;
final CSNGenerator gen = new CSNGenerator(serverId, TimeThread.getTime());
final CSN failing = gen.newCSN();
final CSN parked = gen.newCSN();
final CSN unwound = gen.newCSN();
final List
* A parked change stays owned by the replay thread which parked it while that thread
* goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
* what hands it out again, to whichever thread calls it first once the change it was
* waiting for is gone - the thread which cleared it, as a rule. Changing the number of
* replay threads stops the whole pool and creates another one, so a thread which parked
* a change and went back to the queue is joined while it is idle, and it would end still
* recorded as the owner of that change - a thread which does not exist anymore, while
* every redelivery of a change a replay thread owns is refused as a duplicate. On a
* domain which then goes quiet that change is where this replica's ServerState, and
* every change behind it from every master, stops.
*/
@Test
public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception
{
testSetUp("aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain"));
final DN waitedOn = addEntryForChange("user.986.1");
final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
final DN other = addEntryForChange("user.986.2");
final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
domain.resetUnreplayedChangeAlertThrottle();
final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
"no change of this domain is waiting for another one when this test starts");
final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
final CSN failing = gen.newCSN();
final CSN parked = gen.newCSN();
final CSN applied = gen.newCSN();
final List
* A delivery is dropped rather than queued while the listener thread is down, which it
* is for as long as a recovery is restarting the session, so a change which has to reach
* a replay thread is delivered until it does. A delivery of a change a replay thread
* owns is refused as the duplicate it is, so the deliveries which follow the one that
* was taken cost nothing.
*/
private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain,
final String attributeName, final long expected,
final Supplier delivery, final String message) throws Exception
{
TestTimer timer = new TestTimer.Builder()
.maxSleep(20, SECONDS)
.sleepTimes(500, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
if (getMonitorAttrValue(baseDN, attributeName) != expected)
{
domain.processUpdate(delivery.get());
}
assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message);
}
});
}
/**
* Sets how many replay threads this server runs, the way an administrator would: the
* pool is stopped and created again with that number.
*/
private static void setNumUpdateReplayThreads(int threads) throws Exception
{
assertEquals(TestCaseUtils.applyModifications(true,
"dn: " + SYNCHRO_PLUGIN_DN,
"changetype: modify",
"replace: ds-cfg-num-update-replay-threads",
"ds-cfg-num-update-replay-threads: " + threads), 0,
"the number of replay threads could not be changed");
}
/**
* Puts the number of replay threads back to what this server computes for itself, which
* is what it runs with when the configuration carries no number of its own.
*
* @return the result code of the change, so that a finally can call this without an
* assertion which would replace the failure it is on the way out of
*/
private static int resetNumUpdateReplayThreads() throws Exception
{
return TestCaseUtils.applyModifications(true,
"dn: " + SYNCHRO_PLUGIN_DN,
"changetype: modify",
"delete: ds-cfg-num-update-replay-threads");
}
/** Adds the entry a change of these tests is made on. */
private DN addEntryForChange(String uid) throws Exception
{
return TestCaseUtils.addEntry(
"dn: uid=" + uid + "," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: " + uid,
"cn: Aaccf Amar",
"sn: Amar").getName();
}
/** A delivery of a change whose replay does not run to its end. */
private interface FailingDelivery
{
ModifyMsg newDelivery(CSN csn, DN dn, List
* Read by identity rather than counted where a test is about one thread in particular
* having ended: the pool belongs to the server rather than to a test, so a count says
* whether it is the size it was, not whether the thread which met the error is the one
* which is gone.
*/
private static Set
* It is thrown where the operation is built, which is inside the replay and past the
* point where the change was marked as being replayed by the thread which took it: what
* this pins is the road out of a replay which no {@code catch} of the replay itself used
* to run on.
*/
private static final class ModifyMsgWhoseReplayThrows extends ModifyMsg
{
private final Supplier
* Its operation is built and can not be prepared for its replay, the way
* {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
* change owned by the replay thread. The ack of the delivery is then published in the
* finally every road out of the replay runs through, and this one throws there - which is
* what a session being torn down does.
*/
private static final class ModifyMsgWhoseAckThrows
extends ModifyMsgWhoseOperationRefusesAControl
{
private ModifyMsgWhoseAckThrows(CSN csn, DN dn, List
* {@code ModifyOperationBasis.getEntryDN()} reports INVALID_DN_SYNTAX and returns null
* when the raw DN of the operation does not parse, and the permissive-modify check the
* replay makes before running the operation reads it. Comparing the DN this check looks
* for with what that returns, rather than the other way round, is what keeps this from
* being a NullPointerException thrown before the operation ran: the operation runs,
* reports the syntax of its DN, and the change is stepped over like any other change
* this replica can not apply. Thrown on instead, the change is kept out of the
* ServerState and asked for again for the whole give-up window, with a stack trace per
* delivery, where the syntax of its DN is a verdict on every delivery of it.
*/
@Test
public void aModifyWhoseEntryDNDoesNotParseIsReportedRatherThanThrownOn() throws Exception
{
testSetUp("aModifyWhoseEntryDNDoesNotParseIsReportedRatherThanThrownOn");
logger.error(LocalizableMessage.raw(
"Starting replication test : aModifyWhoseEntryDNDoesNotParseIsReportedRatherThanThrownOn"));
Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.928.1," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.928.1",
"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(19, TimeThread.getTime());
final CSN csn = gen.newCSN();
final List
* The replay fails first - its operation can not be prepared for the replay, the way
* {@code ModifyMsgWhoseOperationRefusesAControl} has it fail - so the change is one this
* replica asks for again, and the ack which says so is where the JVM runs out of memory.
* That is the one throw from there which is not caught: it ends the replay thread, and
* the change is given back on the way out.
*/
private static final class ModifyMsgWhoseAckRunsOutOfMemory
extends ModifyMsgWhoseOperationRefusesAControl
{
private ModifyMsgWhoseAckRunsOutOfMemory(
CSN csn, DN dn, List
* The change is committed before the ack of its delivery is published, and the ack is
* where the JVM runs out of memory: the one throw from there which is not caught, so the
* replay is unwound with the change it was replaying in the data and owned by nobody -
* commit() cleared the owner - and the thread ends on the error. What the give-back on
* the way out of {@code replay()} has left to give back is the changes this thread parked
* as waiting for another one, and the restart it asks for them is the one which is run:
* the road a failed replay takes to ask for its own change again is not on the way.
*
* Nothing on the way in reads what throws here: a message handed to the domain rather
* than published is not one this server acknowledges to anybody.
*/
private static final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
{
private ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
CSN csn, DN dn, List
* The change is committed before the ack of its delivery is published, so this is the
* road on which nothing is owed back to the replication server - and on which the changes
* parked behind this one are still waiting to be handed out.
*/
private static final class AddMsgWhoseAckThrows extends AddMsg
{
private AddMsgWhoseAckThrows(CSN csn, DN dn, String entryUUID, String parentEntryUUID,
Attribute objectClasses, Iterable
* What has its replay fail is not the message but the test which delivers it, through a
* throw at the pre-parse plugin point: an add which is parked behind its parent has to be
* one whose operation builds and runs, or it would be given up on where no operation could
* be built from it. The throw is caught inside the replay, so it is what the replay runs
* once the ack is out - the give-back of the failed change - which reads the CSN off this
* message and is unwound by it.
*/
private static final class AddMsgWhoseReplayIsUnwoundAfterItsAck extends AddMsg
{
private volatile boolean ackPublished;
private AddMsgWhoseReplayIsUnwoundAfterItsAck(CSN csn, DN dn, String entryUUID,
String parentEntryUUID, Attribute objectClasses, Iterable
* The session is left down for ten seconds at the longest between two deliveries, so a
* warning per delivery is the same line every ten seconds for as long as the change is
* retried - and how long that is has been the administrator's to set since #901. The
* budget is unlimited here: this is the domain which must not go silent over a change it
* keeps asking for.
*
* Each warning says how many deliveries were folded into it, and only those: the third
* warning stands for the deliveries since the second one, not for those since the first.
*/
@Test
public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Exception
{
testSetUp("aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long interval = shortenReplayRetryWarningInterval(domain);
try
{
setReplayGiveUpDelay("unlimited");
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = addUserEntry("user.942.retried");
final CSN csn = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
/*
* The session is left down for a second, then two, between the deliveries, so the
* first three of them fall into one interval: one warning, where there were three
* before this one was throttled.
*/
waitForDeliveries(3);
assertEquals(replayRetryWarnings(csn).size(), 1,
"a change which keeps failing must be warned about once per interval, not once per delivery");
/*
* Once the interval has passed the change is warned about again, and the line
* which comes says how many deliveries were folded into no warning meanwhile.
*/
waitForReplayRetryWarnings(csn, 2);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
.as("the warning must say how many failed deliveries it stands for")
.isGreaterThanOrEqualTo(1);
final int deliveriesAtSecondWarning = deliveriesSoFar();
/*
* The deliveries the second warning stands for are not the third one's as well: a
* count which was read rather than taken when the line was written would have every
* warning of an outage count every delivery since its first one. The deliveries
* since the second warning are the one which logged the third and those folded
* into it, so the count is below that number.
*/
waitForReplayRetryWarnings(csn, 3);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(2)))
.as("the third warning must only stand for the deliveries since the second one")
.isLessThan(deliveriesSoFar() - deliveriesAtSecondWarning);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, csn, failed);
throw failed;
}
letTheChangeBeCovered(domain, csn);
}
finally
{
// What can not throw first: a cleanup which throws skips the ones after it.
domain.setReplayRetryWarningInterval(interval);
broker.stop();
resetReplayGiveUpDelay();
}
}
/**
* Test case for [Issue 942]: the deliveries folded into no warning are forgotten when
* this replica stops failing, rather than carried over to the next failure.
*
* The count is what the next warning says it stands for. A warning logged over another
* change, once the backend has served again for a while, must not read as counting the
* deliveries of the failure before it.
*/
@Test
public void aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore() throws Exception
{
testSetUp("aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore");
logger.error(LocalizableMessage.raw("Starting replication test : "
+ "aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long interval = shortenReplayRetryWarningInterval(domain);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry first = addUserEntry("user.942.recovered");
Entry second = addUserEntry("user.942.failing.next");
final CSN csn = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
// Two deliveries which fail: the first is warned about, the second is folded.
broker.publish(new DeleteMsg(first.getName(), csn, getEntryUUID(first.getName())));
waitForDeliveries(2);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, csn, failed);
throw failed;
}
// The backend serves again: the delivery which comes next replays the change.
letTheChangeBeCovered(domain, csn);
/*
* Another change fails, and the interval is not waited out: only the timestamp of the
* throttle is put back, the count is the domain's to keep or to forget.
*/
domain.resetReplayRetryWarningThrottle();
final CSN later = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(second.getName(), later, getEntryUUID(second.getName())));
waitForReplayRetryWarnings(later, 1);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(later).get(0)))
.as("the warning over a new failure must not count the deliveries of the one before")
.isEqualTo(0);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, later, failed);
throw failed;
}
letTheChangeBeCovered(domain, later);
}
finally
{
domain.setReplayRetryWarningInterval(interval);
broker.stop();
}
}
/**
* Test case for [Issue 942]: giving up on the change which was the last one failing
* forgets the deliveries folded into no warning as well.
*
* The line which says the change is being skipped reports every delivery it had, so
* nothing is lost, and the next failure - a day later, on the default budget - must not
* be warned about as if the deliveries of the change given up on were its own.
*/
@Test
public void aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries() throws Exception
{
testSetUp("aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries");
logger.error(LocalizableMessage.raw("Starting replication test : "
+ "aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long interval = shortenReplayRetryWarningInterval(domain);
try
{
setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_OVER_FOLDED_DELIVERIES);
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = addUserEntry("user.942.given.up");
final CSN csn = gen.newCSN();
final CSN later = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
/*
* The change is given up on once its budget is spent, which the ServerState
* covering a change which was never applied says. The budget outlasts the first
* deliveries, so at least one of them was folded into no warning by then: the
* delivery which spends the budget is neither warned about nor folded.
*/
waitUntilCovered(domain, csn);
Assertions.assertThat(deliveriesSoFar())
.as("the budget must have outlasted a delivery which was folded into no warning")
.isGreaterThanOrEqualTo(3);
/*
* Another change fails, and the interval is not waited out: only the timestamp of
* the throttle is put back, the count is the domain's to keep or to forget.
*/
domain.resetReplayRetryWarningThrottle();
broker.publish(new DeleteMsg(tmp.getName(), later, getEntryUUID(tmp.getName())));
waitForReplayRetryWarnings(later, 1);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(later).get(0)))
.as("the warning after a change was given up on must not count its deliveries")
.isEqualTo(0);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, later, failed);
throw failed;
}
/*
* Replayed now that the backend serves again, or given up on: either way it is
* covered. The budget is still the short one here, and nothing spends it: a change is
* only given up on over a delivery which failed, and none fails once the short circuit
* is gone.
*/
letTheChangeBeCovered(domain, later);
}
finally
{
// What can not throw first: a cleanup which throws skips the ones after it.
domain.setReplayRetryWarningInterval(interval);
broker.stop();
resetReplayGiveUpDelay();
}
}
/**
* Test case for [Issue 942]: disabling the domain - for an LDIF import, a restore, or a
* backend being taken offline - forgets the deliveries folded into no warning, along
* with the changes they were deliveries of.
*
* The changes listed as pending do not outlive the ServerState the domain saves on its
* way down, and the recovery from a failed replay goes with them: the first warning
* over the data loaded back must not count the deliveries of a change which is not
* listed anymore.
*/
@Test
public void aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore() throws Exception
{
testSetUp("aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore");
logger.error(LocalizableMessage.raw("Starting replication test : "
+ "aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
final long interval = shortenReplayRetryWarningInterval(domain);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = addUserEntry("user.942.disabled");
final CSN csn = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
/*
* Three deliveries which fail: the first is warned about, the second is folded
* for sure by the time the third is delivered - a delivery is folded once its
* attempts are over, and the next one only comes over the session restarted
* after that.
*/
waitForDeliveries(3);
/*
* The domain goes down and comes back, the way an import or a restore has it: the
* change is not listed anymore, and the replication server sends it again over the
* new session, from the ServerState which was saved without it. The throttle is put
* back while the domain is down, when nothing fails, so that the first failure over
* the data loaded back is warned about straight away - with the count the domain
* kept or forgot.
*/
domain.disable();
domain.resetReplayRetryWarningThrottle();
domain.enable();
waitForReplayRetryWarnings(csn, 2);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
.as("the first warning after the domain was disabled must not count the deliveries before")
.isEqualTo(0);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, csn, failed);
throw failed;
}
letTheChangeBeCovered(domain, csn);
}
finally
{
domain.setReplayRetryWarningInterval(interval);
broker.stop();
}
}
/**
* Test case for [Issue 942]: a change replayed while another one keeps failing does not
* forget the deliveries folded into no warning.
*
* This is what the issue looks like live: one entry which can not be applied here, among
* changes which replay perfectly well. Each of those is a change replayed, and the count
* is forgotten when a change is replayed - only when nothing is failing anymore, though.
* The deliveries folded so far are deliveries of the outage the next warning is about, and
* a warning which said {@code 0 further} over them would be a wrong number which looks
* right.
*/
@Test
public void aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing()
throws Exception
{
testSetUp("aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing");
logger.error(LocalizableMessage.raw("Starting replication test : "
+ "aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
/*
* The interval is left as the server has it, a minute, and only the throttle is put back:
* the second warning must be the one let through below, once the other change has been
* replayed, so that its count says what the domain kept over that replay. A warning the
* interval let through meanwhile would take the count with it.
*/
domain.resetReplayRetryWarningThrottle();
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = addUserEntry("user.942.failing.alone");
final CSN csn = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
/*
* Three deliveries which fail: the first is warned about, the second and the third
* are folded. Two folded at the least, so that a count which was kept is told apart
* from the one delivery which may be folded between the other change being replayed
* and the throttle being put back.
*
* Waited for by the first attempt of the fourth delivery rather than by the last
* attempt of the third: a delivery is folded once its attempts in place are spent,
* after the last of them tripped the short circuit, and the fourth delivery only
* comes over the session restart the third asks for once it has been folded. A
* count read on that last attempt would credit a fold which is not there yet, and
* the throttle put back below before it lands would have the third delivery warned
* about rather than folded - with the one fold there was.
*/
waitForDeliveryAttempts(3 * IN_PLACE_REPLAY_ATTEMPTS + 1);
// Deliveries 2 and 3: delivery 1 was warned about.
final int foldedBeforeTheReplay = 2;
/*
* Another entry is added while the delete keeps failing. An add does not depend on
* the delete of another entry, so it is replayed - a change made while the delete
* is still listed as failing.
*/
final String otherUUID = "94200000-0000-0000-0000-000000000001";
final Entry other = TestCaseUtils.makeEntry(
"dn: uid=user.942.replayed.meanwhile," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.942.replayed.meanwhile",
"cn: Aaccf Amar",
"sn: Amar",
"entryUUID: " + otherUUID);
broker.publish(addMsg(gen, other, otherUUID, baseUUID));
assertNotNull(getEntry(other.getName(), 10000, true),
"the change of another entry must be replayed while the delete keeps failing");
/*
* The backoff of the session restarts is kept under the same guard as the count:
* three deliveries failed, so three restarts ran in a row - the fourth delivery came
* over the third - and the replay of the add forgot none of them.
*/
Assertions.assertThat(domain.getConsecutiveSessionRestarts())
.as("a replay of another change must not forget the backoff of the one still failing")
.isGreaterThanOrEqualTo(3);
/*
* The throttle is put back, so that the next delivery of the delete is warned about
* with the count the domain kept - or forgot - over the replay.
*/
domain.resetReplayRetryWarningThrottle();
waitForReplayRetryWarnings(csn, 2);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
.as("a replay of another change must not forget the deliveries of the one still failing")
.isGreaterThanOrEqualTo(foldedBeforeTheReplay);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, csn, failed);
throw failed;
}
letTheChangeBeCovered(domain, csn);
}
finally
{
broker.stop();
}
}
/**
* Test case for [Issue 942]: giving up on a change while another one keeps failing does
* not forget the deliveries folded into no warning either.
*
* The road a change is given up on when its budget is spent is the road a change no
* operation can be built from takes at its first delivery, and that one needs no budget
* to be waited out: a message whose modifications can not be decoded is given up on while
* the delete which keeps failing is still listed as failing.
*/
@Test
public void givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing()
throws Exception
{
testSetUp("givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing");
logger.error(LocalizableMessage.raw("Starting replication test : "
+ "givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing"));
final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
// The interval is left as the server has it, for the reason the case above gives.
domain.resetReplayRetryWarningThrottle();
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);
Entry tmp = addUserEntry("user.942.failing.alone.too");
Entry other = addUserEntry("user.942.given.up.meanwhile");
final CSN csn = gen.newCSN();
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
try
{
broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
// Three deliveries which fail, two of them folded, waited for as in the case above.
waitForDeliveryAttempts(3 * IN_PLACE_REPLAY_ATTEMPTS + 1);
final int foldedBeforeTheGiveUp = 2;
/*
* A change of another entry which no operation can be built from is given up on at
* its first delivery, while the delete is still listed as failing. The count of the
* changes this replica gave up on says when it has been.
*/
final long givenUpBefore = getMonitorAttrValue(baseDN, "replayed-updates-failed");
broker.publish(
undecodableModifyMsg(gen.newCSN(), other.getName(), getEntryUUID(other.getName())));
assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", givenUpBefore + 1,
"the change which can not be decoded must be given up on while the delete keeps failing");
domain.resetReplayRetryWarningThrottle();
waitForReplayRetryWarnings(csn, 2);
Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
.as("giving up on another change must not forget the deliveries of the one still failing")
.isGreaterThanOrEqualTo(foldedBeforeTheGiveUp);
}
catch (Throwable failed)
{
letTheChangeBeCovered(domain, csn, failed);
throw failed;
}
letTheChangeBeCovered(domain, csn);
}
finally
{
broker.stop();
}
}
/**
* Shortens how long the domain does not warn again about a change it asks for again, and
* returns the interval to put back in a finally: the domain outlives the test methods,
* and none of them can afford the minute of the server.
*
* The throttle is the domain's too: a change another test was retrying less than an
* interval ago would have the first warning of this one folded into its own, so it is put
* back as well.
*
* @param domain the domain of the test
* @return the interval the domain had, in milliseconds
*/
private static long shortenReplayRetryWarningInterval(LDAPReplicationDomain domain)
{
final long interval = domain.getReplayRetryWarningInterval();
domain.setReplayRetryWarningInterval(TEST_REPLAY_RETRY_WARNING_INTERVAL_IN_MS);
domain.resetReplayRetryWarningThrottle();
return interval;
}
/** Adds an entry with the provided uid below the base DN, the entry the change fails on. */
private Entry addUserEntry(String uid) throws Exception
{
return TestCaseUtils.addEntry(
"dn: uid=" + uid + "," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: " + uid,
"cn: Aaccf Amar",
"sn: Amar");
}
/**
* Returns how many deliveries of the delete which can not be replayed have failed since
* the short circuit was registered: a delivery is attempted
* {@link LDAPReplicationDomain#IN_PLACE_REPLAY_ATTEMPTS} times in place before the change
* is asked for again, and each attempt trips the short circuit once.
*
* @return the number of deliveries whose attempts in place are all spent
*/
private static int deliveriesSoFar()
{
return ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")
/ IN_PLACE_REPLAY_ATTEMPTS;
}
/**
* Waits until the delete which can not be replayed has been delivered, and failed, the
* provided number of times.
*
* @param deliveries how many deliveries to wait for
* @throws Exception if the deliveries do not come
*/
private static void waitForDeliveries(final int deliveries) throws Exception
{
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(100, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(deliveriesSoFar() >= deliveries,
"the change was not delivered again after its replay failed: " + deliveriesSoFar()
+ " deliveries where " + deliveries + " were expected");
}
});
}
/**
* Waits until the delete which can not be replayed has been attempted the provided number
* of times, over however many deliveries.
*
* The first attempt of a delivery is one more than the attempts of the deliveries before
* it, and it is what tells that the delivery before it has been warned about or folded:
* a delivery is folded once its attempts in place are spent, after the last of them
* tripped the short circuit, and the next delivery only comes over the session restart
* asked for once it has been. {@link #waitForDeliveries(int)} returns on that last
* attempt, before the fold.
*
* @param attempts how many attempts to wait for
* @throws Exception if the attempts do not come
*/
private static void waitForDeliveryAttempts(final int attempts) throws Exception
{
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(100, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
final int attemptsSoFar =
ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse");
assertTrue(attemptsSoFar >= attempts,
"the change was not attempted again after its replay failed: " + attemptsSoFar
+ " attempts where " + attempts + " were expected");
}
});
}
/**
* Waits until this replica has warned the provided number of times that the provided
* change is being asked for again.
*
* @param csn the CSN of the change whose replay keeps failing
* @param warnings how many warnings to wait for
* @throws Exception if the warnings do not come
*/
private static void waitForReplayRetryWarnings(final CSN csn, final int warnings)
throws Exception
{
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertEquals(replayRetryWarnings(csn).size(), warnings,
"a change which keeps failing was not warned about " + warnings + " times");
}
});
}
/**
* Takes the short circuit back and waits until the change it was failing is covered by
* the ServerState - replayed now that the backend serves again, or given up on - so that
* a case does not leave its change to the next one: a change left failing here would be
* the next test's, holding its ServerState back, its session restart backoff up and the
* warnings of that test folded into its own.
*
* Called at the end of a case, and from its catch with the failure when it has one,
* rather than from a finally: a wait which expired in a finally would replace the
* assertion it was cleaning up after.
*
* @param domain the domain of the test
* @param csn the CSN of the change
* @throws Exception if the change is not covered
*/
private static void letTheChangeBeCovered(final LDAPReplicationDomain domain, final CSN csn)
throws Exception
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
waitUntilCovered(domain, csn);
}
/**
* {@link #letTheChangeBeCovered(LDAPReplicationDomain, CSN)} for a case which failed: the
* failure is what is thrown, and what went wrong here is added to it.
*
* @param domain the domain of the test
* @param csn the CSN of the change
* @param failed the failure of the case
*/
private static void letTheChangeBeCovered(
final LDAPReplicationDomain domain, final CSN csn, final Throwable failed)
{
try
{
letTheChangeBeCovered(domain, csn);
}
catch (Throwable late)
{
failed.addSuppressed(late);
}
}
/**
* Waits until the ServerState of the domain covers the provided change: it was replayed
* once the backend served again, or given up on.
*
* @param domain the domain of the test
* @param csn the CSN of the change
* @throws Exception if the change is not covered
*/
private static void waitUntilCovered(final LDAPReplicationDomain domain, final CSN csn)
throws Exception
{
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 replayed once the backend serves again, or given up on");
}
});
}
/**
* Returns how many deliveries the provided warning says were folded into no warning of
* their own.
*
* @param warning a warning about a change being asked for again
* @return the count the warning carries
*/
private static int foldedDeliveriesSaidBy(String warning)
{
final Matcher count = Pattern.compile("(\\d+) further deliveries").matcher(warning);
assertTrue(count.find(),
"the warning does not say how many deliveries it stands for: " + warning);
return Integer.parseInt(count.group(1));
}
/**
* Returns the warnings this replica logged about the provided change being asked for
* again, oldest first.
*
* The error log of the test server is written to a writer which keeps every record, so
* the warnings about one change are the records which carry the ordinal of the message
* and the CSN of the change.
*
* The test server registers two error log publishers over that one writer, so it keeps
* every record twice, each copy timestamped by its publisher: a warning is two records
* which read the same once the timestamp is left out - the two publishers read the clock
* one after the other, and a second which turns over between the two reads would have
* one warning counted as two by the records as they are. So what is returned here is the
* messages which differ, each as many times as half its records: two warnings which read
* the same are two warnings, which happens when the domain was disabled in between and
* asks for the change again as one it does not remember.
*
* @param csn the CSN of the change whose replay keeps failing
* @return the warnings which name it, in the order they were first logged
*/
private static 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 replay is unwound once the ack of its delivery is out.
*
* Its operation is built and can not be prepared for its replay, the way
* {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
* change owned by the replay thread. The CSN of the change is then read again - by the
* road which gives it back and asks for it again - and it is that read which throws here:
* past the ack, past the finally it is published in, and past every catch the replay
* itself has. So the only thing left to give the change back is the road out of
* {@code replay()}.
*/
private static final class ModifyMsgWhoseReplayIsUnwoundAfterItsAck
extends ModifyMsgWhoseOperationRefusesAControl
{
private volatile boolean ackPublished;
private ModifyMsgWhoseReplayIsUnwoundAfterItsAck(
CSN csn, DN dn, List
* The operation is built - so the replay is past the point where a message is given up
* on - and the list of request controls it carries can not be added to, so the
* ManageDsaIT control the replay puts on every operation throws before
* {@code OperationContext.getCSN(op)} is reached. Such a message can not travel the
* protocol: {@code ModifyMsg.createOperation()} builds an operation whose controls can
* be added to, so this one is handed to the domain rather than published.
*/
private static class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
{
private ModifyMsgWhoseOperationRefusesAControl(
CSN csn, DN dn, List
* {@code ModifyOperationBasis.getEntryDN()} reports INVALID_DN_SYNTAX and returns null
* for such an operation, which is what the permissive-modify check of the replay reads.
* Such a message can not travel the protocol: the DN of a ModifyMsg is parsed 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 or first inside it - the reset then runs on an attribute which is there to be
* removed, or finds it gone already, which it takes for the default it was asking for.
*
* @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