* The thread which gave them back runs it itself, so that the delivery they wait for is * not left to the next tick of the state checkpointer - unless that thread is stopping, in * which case it leaves the request standing the way a stopping thread leaves the request it * makes for the change it abandons: the threads of the pool are stopped one after the * other and joined, and each running a restart on its way out would have the configuration * change which is stopping them wait for one restart per thread. *
* The replay runs on the thread of the test, which is what says whether that thread is
* stopping, and the error which unwinds it is caught here rather than ending a replay
* thread. The thread which is stopping is stopped by the ack of the change it applied, and
* the case asserts that the change was applied: stopped before the replay, it abandons the
* change unapplied at the top of its first attempt, and the abandon road asks for a restart
* of its own next to the give-back's, which would then be pinned by nothing - that road is
* the third case's, which pins the abandon arm of the catch. The restart is asked to fail
* once, so that the thread which ran it is the one which met the failure: a replay thread
* carries the failure out with the error it is ending on, the state checkpointer reports
* it. The restart is asked for at once, without the backoff, and the count of the restarts
* in a row says so: only the restart run again after the failure waits its backoff out.
*/
@SuppressWarnings("javadoc")
public class ParkedChangeGiveBackTest extends ReplicationTestCase
{
private static final int RS_ID = 612;
private static final int DS_ID = 1;
private static final AtomicBoolean RUNNING = new AtomicBoolean(false);
/**
* How long the state checkpointer is given to run a request left standing and report
* the failure it was asked to meet: its next tick, at most a second away, and the
* report.
*/
private static final long CHECKPOINTER_BOUND_IN_MS = 5000;
/**
* How long the session is given to come back once a restart failed: the backoff a
* restart which follows a failed one is owed - the second in a row here, two seconds -
* and the tick of the state checkpointer which runs it.
*/
private static final long RESTART_BOUND_IN_MS = 10000;
private DN baseDN;
private ReplicationServer replicationServer;
private LDAPReplicationDomain domain;
private TestSynchronousReplayQueue queue;
private CSNGenerator gen;
@BeforeMethod
public void setUpLocal() throws Exception
{
baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
final int rsPort = TestCaseUtils.findFreePort();
replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
rsPort, "parkedChangeGiveBackTestDb", 0, RS_ID, 0, 100, new TreeSet
* The thread is stopped by the ack of the change it applied, so the change is committed
* and owned by nobody by the time the replay is unwound, and the request the give-back
* makes for the parked change is the one request which stands for the checkpointer. The
* case asserts that road: a thread stopped before the replay hands its change back on a
* road of its own, which asks for a restart next to the give-back's and would stand in
* for it.
*/
@Test(timeOut = 120_000)
public void aStoppingThreadLeavesTheRestartItAskedForToTheStateCheckpointer() throws Exception
{
parkAChangeBehindABarrier(addEntry("waitedOn"));
final Entry other = addEntry("other");
final int reportedBefore = restartFailureReports().size();
final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
final int restartsBefore = domain.getConsecutiveSessionRestarts();
domain.failNextSessionRestarts(1);
final OutOfMemoryError unwinding = unwindAReplay(other, Stopped.BY_THE_ACK);
assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore + 1,
"the change must be applied and its replay unwound by the ack, not abandoned unapplied");
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
"the change parked by the replay which was unwound must be given back");
assertThat(injectedRestartFailuresAmong(unwinding.getSuppressed()))
.as("a thread which is stopping must not run the restart it asked for")
.isEmpty();
final long deadline = System.currentTimeMillis() + CHECKPOINTER_BOUND_IN_MS;
while (restartFailureReports().size() == reportedBefore)
{
assertTrue(System.currentTimeMillis() < deadline, "the state checkpointer did not run"
+ " the restart the stopping thread left standing: the failure that restart was"
+ " asked to meet was never reported");
Thread.sleep(50);
}
awaitConnected(RESTART_BOUND_IN_MS,
"the session was not brought back after the restart which failed");
assertEquals(domain.getConsecutiveSessionRestarts(), restartsBefore + 1,
"the restart the give-back asked for must be run at once, without the backoff: the one"
+ " run again after the failure is the one which waits it out");
}
/**
* A thread stopped before its first attempt hands back the change it did not apply, the
* way a stopping thread abandons a replay: the ack of an abandoned change is published
* all the same, and it is what runs out of memory here, so the abandon road taken is the
* one of the catch - the change is still this thread's when the replay is unwound.
*/
@Test(timeOut = 120_000)
public void aStoppingThreadHandsBackTheChangeItDidNotApply() throws Exception
{
parkAChangeBehindABarrier(addEntry("waitedOn"));
final Entry other = addEntry("other");
final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
final CSN abandoned = gen.newCSN();
unwindAReplay(other, abandoned, Stopped.BEFORE_THE_REPLAY);
assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore,
"a thread stopped before the replay must not apply the change");
assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), abandoned))
.as("a stopping thread must hand back the change it did not apply")
.isNotEmpty();
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
"the change parked by the thread which is stopping must be given back");
}
private void awaitConnected(long boundInMs, String message) throws Exception
{
final long deadline = System.currentTimeMillis() + boundInMs;
while (!domain.isConnected())
{
assertTrue(System.currentTimeMillis() < deadline, message);
Thread.sleep(50);
}
}
private Entry addEntry(String cn) throws Exception
{
return TestCaseUtils.addEntry(
"dn: cn=" + cn + "," + TEST_ROOT_DN_STRING,
"objectClass: top",
"objectClass: person",
"cn: " + cn,
"sn: " + cn);
}
/**
* Replays, on the thread of this test, a change whose replay fails and stays listed - its
* operation is built and then refused, so it is asked for again rather than stepped over -
* and then a change on the same entry, which is parked as waiting for it and owned by this
* thread from then on. The restart the failed change asks for is run on this thread as
* well, so the session is back once this returns.
*/
private void parkAChangeBehindABarrier(Entry entry) throws Exception
{
final String entryUUID = getEntryUUID(entry.getName());
final CSN failing = gen.newCSN();
replayMsg(new ModifyMsgWhoseOperationRefusesAControl(failing, entry.getName(),
generatemods("description", "the replay of this change fails"), entryUUID), RUNNING);
assertFalse(domain.getServerState().cover(failing),
"the change whose replay fails must stay listed as one which is not in the data");
awaitConnected(RESTART_BOUND_IN_MS,
"the session was not brought back for the change whose replay failed");
replayMsg(new ModifyMsg(gen.newCSN(), entry.getName(),
generatemods("description", "the change which was parked as a dependency"), entryUUID),
RUNNING);
assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
"a change which waits for one that is not in the data must be parked");
}
/** When the thread a replay runs on is stopped, if it is. */
private enum Stopped
{
/** It is running throughout. */
NEVER,
/** By the ack of the change it applied: the change is committed, and the replay is unwound. */
BY_THE_ACK,
/**
* Before the replay: the change is abandoned unapplied at the top of its first attempt,
* and the ack which says so is what runs out of memory.
*/
BEFORE_THE_REPLAY
}
private OutOfMemoryError unwindAReplay(Entry entry, Stopped stopped) throws Exception
{
return unwindAReplay(entry, gen.newCSN(), stopped);
}
/**
* Replays, on the thread of this test, a change whose ack runs out of memory, and returns
* the error the replay was unwound on.
*
* @param stopped when the thread the replay runs on is stopped: running until the ack,
* the change is applied rather than abandoned at the top of its first attempt
*/
private OutOfMemoryError unwindAReplay(Entry entry, CSN csn, Stopped stopped) throws Exception
{
// Fresh for every replay: the ack is what sets it, and nothing puts it back.
final AtomicBoolean stopping = new AtomicBoolean(stopped == Stopped.BEFORE_THE_REPLAY);
try
{
replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(csn, entry.getName(),
generatemods("description", "the replay of this change is unwound by its ack"),
getEntryUUID(entry.getName()), stopped == Stopped.BY_THE_ACK ? stopping : null), stopping);
}
catch (OutOfMemoryError unwinding)
{
// The error is the fixture's own, and this is the thread it would have ended.
return unwinding;
}
throw new AssertionError("the replay was not unwound: the ack of the delivery must run out of memory");
}
/** The throwables among the provided ones which a restart threw because a test asked it to. */
private static List