From c6b483127f13fc7d96dbc90c5157ab113ac12973 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 16 Sep 2026 10:07:08 +0000
Subject: [PATCH] [#925] Keep asking for a session restart until it has run (#981)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 267 +++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 267 insertions(+), 0 deletions(-)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
index 77d0acf..d57e0f9 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
@@ -156,6 +156,20 @@
*/
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.
+ * <p>
+ * 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;
@@ -2355,6 +2369,259 @@
}
/**
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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<String> records = errorLogRecordsOf(() -> {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ assertNull(getEntry(tmp.getName(), 60000, false),
+ "the change was not delivered again after the session restarts which failed");
+ return null;
+ });
+ assertEquals(domain.getSessionRestartFailuresLeft(), 0,
+ "the restarts which were asked to fail never ran, so this test proves nothing");
+ /*
+ * The failure the replay thread meets is reported by the replay thread's own catch,
+ * as an exception replaying a message; only the checkpointer says this, once per
+ * restart which threw on it, and one did. The replay thread's request is its own to
+ * run unless the checkpointer's tick lands in the instants between the request
+ * being made and being taken: the checkpointer then spends both failures and says
+ * this twice, which is the same report on a rarer road rather than a double one. A
+ * count of none has no road - the replay thread's own failure is never reported as
+ * this - so the count still says that the checkpointer reports what threw on it.
+ */
+ final int reported = countRecordsOf(records,
+ "Could not restart the replication session of domain \"" + baseDN + "\"");
+ assertTrue(reported == 1 || reported == 2, "the state checkpointer reports the"
+ + " restart which threw on it once, or twice when its tick took the replay"
+ + " thread's request as well, and runs it again: " + reported + " in " + records);
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1,
+ "the change must be recorded as replayed");
+ }
+ finally
+ {
+ domain.failNextSessionRestarts(0);
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 925]: a session restart which was asked for without the backoff
+ * and could not run is asked for again with it. A replay thread on its way out - the
+ * number of them is being changed - hands its change back and asks for the restart
+ * without the wait, since the backend is not what is going away; the state checkpointer
+ * runs what it asked for. A session which can not be started is what the wait exists
+ * for, though, and a request given back as it was made would have the checkpointer stop
+ * and fail to start the session once a second, for as long as the failure lasts.
+ * <p>
+ * 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.
+ * <p>
+ * 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<Throwable> reconfigurationFailure = new AtomicReference<>();
+ Thread reconfiguration = null;
+ boolean reconfigurationFinished = true;
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+ parked.awaitParked(60, SECONDS);
+
+ reconfiguration = startReplayThreadReconfiguration(2, reconfigurationFailure);
+ awaitStoppingTheReplayThreads(reconfiguration, reconfigurationFailure);
+
+ /*
+ * The thread which holds the change has been asked to stop: let go of it, and it
+ * hands the change back and asks for the restart on its way out. The restart is
+ * the checkpointer's to run, and it fails where it would start the session again.
+ */
+ domain.failNextSessionRestarts(1);
+ final List<String> records = errorLogRecordsOf(() -> {
+ parked.release(ResultCode.UNAVAILABLE.intValue());
+
+ final long deadline = System.nanoTime() + SECONDS.toNanos(30);
+ while (domain.getSessionRestartFailuresLeft() > 0)
+ {
+ assertTrue(System.nanoTime() < deadline,
+ "the session restart which was asked to fail did not run within 30 s");
+ Thread.sleep(50);
+ }
+ Thread.sleep(INTO_THE_BACKOFF_IN_MS);
+ assertFalse(domain.isConnected(), "the session was started back within "
+ + INTO_THE_BACKOFF_IN_MS + " ms of the restart which threw: the request a"
+ + " replay thread on its way out made without the backoff was given back"
+ + " without it, and the checkpointer ran it again on its next tick rather"
+ + " than one backoff later");
+ return null;
+ });
+ /*
+ * Which thread ran the restart is read off the error log rather than off the clock:
+ * only the state checkpointer reports a restart which threw on it as this, where a
+ * replay thread reports the failure it meets as an exception replaying a message.
+ * One report says the checkpointer ran the restart the thread asked for; none would
+ * say the thread ran it itself on its way out, as it did before, and a second has no
+ * road - one failure was left to spend.
+ */
+ final int reported = countRecordsOf(records,
+ "Could not restart the replication session of domain \"" + baseDN + "\"");
+ assertEquals(reported, 1, "the restart a replay thread on its way out asked for was"
+ + " run by that thread rather than left to the state checkpointer: " + reported
+ + " in " + records);
+
+ /*
+ * Stop parking before the session is back: the delivery it brings is the change
+ * being delivered again, and it is to be applied. The replication server owns the
+ * change the thread handed back, so the session started once the backoff is out
+ * is what has it delivered.
+ */
+ parked.deregister();
+ reconfiguration.join(SECONDS.toMillis(60));
+ assertFalse(reconfiguration.isAlive(),
+ "the replay threads were reconfigured, but applyConfigurationChange never returned");
+ if (reconfigurationFailure.get() != null)
+ {
+ throw new AssertionError("the replay threads could not be reconfigured",
+ reconfigurationFailure.get());
+ }
+ assertNull(getEntry(tmp.getName(), 60000, false),
+ "the change was not delivered again once the backoff was out");
+ }
+ finally
+ {
+ /*
+ * Whatever happened above: no failure may be left to inject, no replay thread of
+ * this server may be left parked, and the reconfiguration has to be over before
+ * the pool is restored - see aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain
+ * for why the pool is otherwise left alone.
+ */
+ domain.failNextSessionRestarts(0);
+ parked.deregister();
+ if (reconfiguration != null)
+ {
+ reconfiguration.join(SECONDS.toMillis(60));
+ reconfigurationFinished = !reconfiguration.isAlive();
+ }
+ if (reconfigurationFinished)
+ {
+ setNumberOfReplayThreads(null);
+ }
+ }
+ assertTrue(reconfigurationFinished,
+ "the replay thread reconfiguration never finished; the pool was left alone");
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
* Test case for [Issue 889]: the result code the server puts on an internal error is
* configurable and only has to report a failure, so it can be set to one conflict
* resolution knows how to solve. Such a change is left to conflict resolution, and when
--
Gitblit v1.10.0