* The export of a total update publishes its entries over the session of the domain, from a * thread of the export pool, while the replay of the domain keeps running. A change which can * not be replayed meanwhile is given back for the replication server to send again, and that * takes a session restart. Run by the thread which released the change, the restart stops the * broker the export publishes over, and {@code exportLDIFEntry()} gives the export up on it: the * replica being initialized is left to be initialized again, for a change which would have * waited (issue #1048). The restart has to wait for the export instead, and the state * checkpointer runs it once the export is over. *
* The importer is a broker of this test, so that the test says when the export moves: the * exporter publishes no more than the initialization window ahead of the importer's * acknowledgements, and the change is replayed while the export waits for one. */ @SuppressWarnings("javadoc") public class ReplayDuringExportTest extends ReplicationTestCase { /** * A total update needs a backend which keeps its data across the export, and one which * the exporter can lock: the {@code userRoot} backend, as for the import direction. */ private static final String EXAMPLE_DN = "dc=example,dc=com"; private static final int RS_ID = 612; private static final int DS_ID = 1; private static final int IMPORTER_ID = 2; /** How many entry messages the exporter publishes ahead of the importer's acknowledgements. */ private static final int INIT_WINDOW = 2; /** * An entry message carries a buffer of the export stream rather than one entry, so the data * has to outgrow the window by that much before the exporter waits for an acknowledgement. */ private static final int ENTRY_MSG_BYTES = 8192; private static final int BULK_ENTRY_BYTES = 4096; private static final int BULK_ENTRIES = 2 * (INIT_WINDOW + 2); private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false); /** * How long the export is given to release its context once the importer has left the full * update status - the exporter waits for that status to go, and releases it then. *
* Short enough that the receive of the stream and this wait fit inside the timeout of the
* case with room to spare: a case which ends on the timeout prints none of the messages
* which say what went wrong.
*/
private static final long EXPORT_END_BOUND_IN_MS = 30_000;
/**
* How long the session is given to come back once a restart has been run for a change which
* was given back: the wait that restart is owed, and the start of the session.
*/
private static final long SESSION_BACK_BOUND_IN_MS = 30_000;
private DN baseDN;
private ReplicationServer replicationServer;
private LDAPReplicationDomain domain;
private TestSynchronousReplayQueue queue;
private ReplicationBroker importer;
private CSNGenerator gen;
@BeforeMethod
public void setUpLocal() throws Exception
{
baseDN = DN.valueOf(EXAMPLE_DN);
TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN);
final int rsPort = TestCaseUtils.findFreePort();
replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
rsPort, "replayDuringExportTestDb", 0, RS_ID, 0, 100, new TreeSet
* The attempts in place are spent - the backend is live, an export takes nothing away - and
* the change is given back and asked for again, as it is when nothing else is going on: what
* waits is the session restart that takes. The restart stands as a request for as long as
* the export runs, and the state checkpointer, which holds its own restarts back for the
* same reason, runs it when the export is over. Without the hold the replay thread stops the
* broker the exporter publishes over: the export ends on the entries which had been
* published, with {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT}, the rest never reaches
* the importer, and the importer has to be initialized again.
*/
@Test(timeOut = 120_000)
public void aReplayWhichFailsDuringTheExportLeavesTheSessionToTheExport() throws Exception
{
final Entry entry = TestCaseUtils.addEntry(
"dn: cn=renamedSince," + EXAMPLE_DN,
"objectClass: top",
"objectClass: person",
"cn: renamedSince",
"sn: renamedSince");
final String entryUUID = getEntryUUID(entry.getName());
final Entry foldedInto = addPersonEntry("folded");
addEntriesWorthMoreThanTheWindow();
final long exportedEntries = countEntriesOfTheDomain();
/*
* The change goes through the replication server, which is what has it to deliver again
* once the session has been restarted for it; the replay queue of the domain is the
* test's, so the change is replayed when the test says, which is during the export.
*/
final CSN csn = gen.newCSN();
importer.publish(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
generatemods("description", "replayed during the export"), entryUUID));
final LDAPUpdateMsg delivered = awaitDelivery(csn, 30_000, "the change was not delivered");
startExport();
final List
* The replay is unwound by the ack of a change it had applied, which is the road the give-back
* of the parked changes is reached from (issue #954): what that road hands back is reported as
* given back to a replication server "which still owns it and sends it again", and during an
* export it does not send it yet. These changes are handed to the domain rather than published,
* so the redelivery the restart brings is the case above's to assert; what is asserted here is
* that the export is not cut for them and that their wait is reported.
*
* The same road is walked once before the export, which is the negative arm of the line: the
* restart of that give-back runs, so nothing of it waits and nothing says it does.
*
* The fixture is the shape {@code ParkedChangeGiveBackTest} gives that road - a change parked
* behind one whose operation is refused, and a replay unwound by an ack - over the backend
* this case exports, and with a total update running over the session.
*/
@Test(timeOut = 120_000)
public void theParkedChangesGivenBackDuringTheExportAreReportedAsHeld() throws Exception
{
final Entry waitedOn = addPersonEntry("waitedOn");
final Entry unwoundBeforeTheExport = addPersonEntry("unwoundBefore");
final Entry unwoundDuringTheExport = addPersonEntry("unwoundDuring");
final CSN parkedBeforeTheExport = parkAChangeBehindABarrier(waitedOn);
final long generationBefore = sessionGeneration();
unwindTheReplayOf(unwoundBeforeTheExport);
/*
* Read the moment the replay returns: the restart of this give-back is run by this thread
* before the replay returns, and one left standing would be run by the state checkpointer
* within its tick, where awaitConnected() below could not tell the two apart - it would
* find the session up either way, restarted or never stopped.
*/
assertThat(sessionGeneration())
.as("the thread which gave the parked change back did not run the restart it asked"
+ " for: no total update is being processed, so nothing holds it")
.isGreaterThan(generationBefore);
assertThat(errorLogRecordsOf(
NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK.ordinal(), parkedBeforeTheExport))
.as("the change parked by the replay which was unwound was not given back")
.isNotEmpty();
assertThat(errorLogRecordsOf(
NOTE_REPLAY_SESSION_RESTART_HELD_BY_TOTAL_UPDATE.ordinal(), parkedBeforeTheExport))
.as("the restart of the give-back was reported as held while no total update was being"
+ " processed: it ran")
.isEmpty();
awaitConnected("the session was not brought back by the restart the give-back ran");
final CSN parked = parkAChangeBehind(waitedOn);
addEntriesWorthMoreThanTheWindow();
final long exportedEntries = countEntriesOfTheDomain();
startExport();
final List
* The DN it carries is not in the data and its entryUUID is the provided entry's, which is
* what has the replay look the entry up by that UUID - the search which is short-circuited.
*
* @return the CSN of the change whose replay failed
*/
private CSN failAReplayOf(Entry entry) throws Exception
{
final CSN csn = gen.newCSN();
// The registration counts the searches it refuses from zero, so the count below is this
// replay's own.
ShortCircuitPlugin.registerShortCircuit(
OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
try
{
replayHandedOver(new ModifyMsg(csn, DN.valueOf("cn=alsoMovedAway," + EXAMPLE_DN),
generatemods("description", "replayed during the export as well"),
getEntryUUID(entry.getName())));
// Read before the short circuit is deregistered, which forgets the count with it.
assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse"))
.as("the replay of this change must have failed: every attempt in place makes the"
+ " entryUUID search which does not run")
.isGreaterThanOrEqualTo(LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS);
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
}
return csn;
}
private static Entry addPersonEntry(String cn) throws Exception
{
return TestCaseUtils.addEntry(
"dn: cn=" + cn + "," + EXAMPLE_DN,
"objectClass: top",
"objectClass: person",
"cn: " + cn,
"sn: " + cn);
}
/**
* Replays, on the thread of this test, a change whose operation is refused - it stays listed
* as one which is not in the data, and every change on that entry waits for it - and then a
* change which is parked behind it.
*
* @return the CSN of the change which is left parked
*/
private CSN parkAChangeBehindABarrier(Entry entry) throws Exception
{
final CSN failing = gen.newCSN();
replayHandedOver(new ModifyMsgWhoseOperationRefusesAControl(failing, entry.getName(),
generatemods("description", "the replay of this change fails"),
getEntryUUID(entry.getName())));
assertFalse(domain.getServerState().cover(failing),
"the change whose replay fails must stay listed as one which is not in the data");
awaitConnected("the session was not brought back for the change whose replay failed");
return parkAChangeBehind(entry);
}
/**
* Replays, on the thread of this test, a change on an entry whose barrier is still missing
* from the data: it is parked as waiting for that one and owned by this thread from then on,
* since nothing hands a parked change out again while what it waits for is missing
* (issue #954).
*
* @return the CSN of the change which is left parked
*/
private CSN parkAChangeBehind(Entry entry) throws Exception
{
final CSN parked = gen.newCSN();
replayHandedOver(new ModifyMsg(parked, entry.getName(),
generatemods("description", "the change which waits for the one which failed"),
getEntryUUID(entry.getName())));
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
"a change which waits for one that is not in the data must be parked");
return parked;
}
/**
* Reads the generation of the session of the domain, which every stop and start of it bumps,
* under {@code serviceStateLock}, as {@code getSessionGeneration()} asks.
*/
private long sessionGeneration() throws Exception
{
final Field lockField = ReplicationDomain.class.getDeclaredField("serviceStateLock");
lockField.setAccessible(true);
final Method getSessionGeneration =
ReplicationDomain.class.getDeclaredMethod("getSessionGeneration");
getSessionGeneration.setAccessible(true);
synchronized (lockField.get(domain))
{
return (Long) getSessionGeneration.invoke(domain);
}
}
/** Waits for the session of the domain to be up, which a restart leaves it. */
private void awaitConnected(String orElse) throws Exception
{
final long deadline = System.currentTimeMillis() + SESSION_BACK_BOUND_IN_MS;
while (!domain.isConnected())
{
assertTrue(System.currentTimeMillis() < deadline, orElse);
Thread.sleep(50);
}
}
/**
* Replays, on the thread of this test, a change whose ack runs out of memory once it is
* applied: the replay is unwound with that change in the data and owned by nobody, so what
* the give-back on the way out of {@code replay()} has to hand back is what this thread
* parked.
*/
private void unwindTheReplayOf(Entry entry) throws Exception
{
try
{
replayHandedOver(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(gen.newCSN(),
entry.getName(), generatemods("description", "the replay of this change is unwound"),
getEntryUUID(entry.getName())));
}
catch (OutOfMemoryError unwound)
{
// The error is the fixture's own, and this is the thread it would have ended.
return;
}
throw new AssertionError(
"the replay was not unwound: the ack of the delivery must run out of memory");
}
/**
* Hands a change to the domain rather than publishing it, and replays it on the thread of
* this test: what the replication server has to deliver again is the published change of the
* case above, and these are the changes whose give-back this one is about.
*/
private void replayHandedOver(UpdateMsg msg) throws Exception
{
domain.processUpdate(msg);
replay(queue.take().getUpdateMessage());
}
/** Adds entries whose export outgrows the initialization window, so that the exporter waits. */
private void addEntriesWorthMoreThanTheWindow() throws Exception
{
assertThat(BULK_ENTRIES * BULK_ENTRY_BYTES)
.as("the data must outgrow the window for the exporter to wait for an acknowledgement")
.isGreaterThan((INIT_WINDOW + 1) * ENTRY_MSG_BYTES);
final char[] padding = new char[BULK_ENTRY_BYTES];
Arrays.fill(padding, 'x');
for (int i = 0; i < BULK_ENTRIES; i++)
{
TestCaseUtils.addEntry(
"dn: cn=bulk" + i + "," + EXAMPLE_DN,
"objectClass: top",
"objectClass: person",
"cn: bulk" + i,
"sn: bulk" + i,
"description: " + new String(padding));
}
}
private long countEntriesOfTheDomain() throws Exception
{
return getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN)
.getNumberOfEntriesInBaseDN(baseDN);
}
/**
* Has the importer ask this replica for a total update, and returns once the export has
* begun: the {@code InitializeTargetMsg} which starts it has arrived.
*/
private void startExport() throws Exception
{
// The export is refused while this replica does not see the importer in its topology.
final long deadline = System.currentTimeMillis() + 30_000;
while (!domain.getReplicaInfos().containsKey(IMPORTER_ID))
{
assertTrue(System.currentTimeMillis() < deadline,
"the domain did not see the importer in its topology");
Thread.sleep(20);
}
importer.publish(new InitializeRequestMsg(baseDN, IMPORTER_ID, DS_ID, INIT_WINDOW));
// The exporter waits for the importer to be in the full update status before it streams.
importer.signalStatusChange(ServerStatus.FULL_UPDATE_STATUS);
final ReplicationMsg msg = receiveTotalUpdateMsg(30_000);
assertThat(msg).as("the total update did not begin").isInstanceOf(InitializeTargetMsg.class);
}
/**
* Receives entry messages up to the window and acknowledges none of them: the exporter
* publishes no more than the window ahead of the last acknowledgement, so its next entry
* message waits for one from now on.
*/
private List
* An export which does not end all the same is printed rather than asserted: it is what
* the assertions which follow this call wait for - the change the restart brings back once
* the export is over - and a throw out of this {@code finally} would replace the failure
* of the stream above it, which is the one worth reading.
*/
private void finishExport(List