+ * The request outlives the thread which made it and the restart which could not run it, + * so that no change is left waiting for a delivery nobody asks for: the restart is run + * by whichever thread takes the request, and by {@link ServerStateFlush} when no thread + * comes back for it. + */ + private final SessionRestartRequests sessionRestarts = new SessionRestartRequests(); + /** + * Woken when a session restart which is sitting through its backoff has nothing left to + * wait for: the domain it would start a session for is going away or is being disabled. */ - private final AtomicBoolean sessionRestartRequested = new AtomicBoolean(); + private final Object sessionRestartBackoff = new Object(); + /** + * How many times {@link #sessionRestartBackoff} has been woken, so that a wake is never + * missed. + *
+ * The wake is given after the flag it stands for is set, and the wait checks the flag + * again once woken - but {@code disabled} does not stay set: an {@link #enable()} which + * runs between the wake and that check puts it back, and the wait would go on for the + * rest of the backoff, with the session {@code enable()} started left to be found gone + * at its end. The count only grows, so a restart which took it where it stopped the + * session sees every wake given since, whether it was waiting yet or not. + */ + @GuardedBy("sessionRestartBackoff") + private long sessionRestartBackoffWakes; /** * How many times in a row the session was restarted without a change being replayed in * between. The backoff is computed from this rather than from the failures of the @@ -368,6 +393,13 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * shortest for as long as the outage lasts. */ private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger(); + /** + * How many of the next session restarts must fail rather than start the session again. + *
+ * Only there for the tests, which have no other way to fail a restart: see {@link + * #failNextSessionRestarts(int)}. It is zero in production, where nothing ever sets it. + */ + private final AtomicInteger sessionRestartFailuresToInject = new AtomicInteger(); /** * Set by {@link #restartService()} when it left the session of this domain alone, so * that the configuration change which asked for the restart can say so. @@ -632,7 +664,8 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { /** * The thread that periodically saves the ServerState of this - * LDAPReplicationDomain in the database. + * LDAPReplicationDomain in the database, and runs the session restart this domain has + * been asked for when the threads which asked for it are gone or could not run it. */ private class ServerStateFlush extends DirectoryThread { @@ -656,6 +689,12 @@ public void run() { saveState(); } + /* + * Run outside the monitor above, as the save is: a session restart holds the + * backoff a failing backend is owed, and shutdown() takes that monitor to wake + * this thread up. + */ + runPendingSessionRestart(); } catch (InterruptedException e) { @@ -2572,6 +2611,13 @@ public void shutdown() { flushThread.notifyAll(); } + /* + * A session restart which is sitting through its backoff has nothing left to wait + * for either, and this thread waits for the checkpointer below: a restart left to + * wait out a backend which is not what is going away would hold this shutdown for + * as long as the backoff it had climbed to. + */ + wakeSessionRestartBackoff(); DirectoryServer.deregisterAlertGenerator(this); getServerContext().getBackendConfigManager() @@ -2712,8 +2758,12 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * its own road: the change is given back counted, the way every other unwound * replay gives it back, but the line which says it is being asked for again is * not built - that asks the JVM for the memory it has just refused - and the - * session is restarted without sitting through the backoff, since the thread - * which is doing it is on its way out. + * restart is asked for without the backoff, since the thread which runs it is + * on its way out. That holds for the restart as first run: one which throws is + * given back with the backoff, as any restart which could not run is, and the + * last resort below runs what is standing - the given-back request, merged with + * its own - on this same thread, backoff and all. Only a restart which throws + * there as well is left to the state checkpointer. * * A change whose budget is spent is still reported and still raises its alert * on this road: it is the one line which says this replica has diverged, and an @@ -2732,14 +2782,14 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * of memory - would leave the change owned by this thread after all. Hand it back * bare, without the failure count that road did not reach, and restart the session * so that it is delivered again: this is the last resort, the throwable is rethrown - * whatever happens here, and the thread this runs on may well be ending on it - so - * a request left for the next recovery of this domain to pick up is a request which - * may never be run. + * whatever happens here, and the thread this runs on may well be ending on it. A + * restart which can not run here leaves its request standing, and the state + * checkpointer of this domain runs it. */ if (owned != null) { remotePendingChanges.replayFailed(owned); - sessionRestartRequested.set(true); + sessionRestarts.request(SessionRestart.NOW); /* * Reported and restarted under guards of their own, and in that order: the report * is the line an operator acts on, and the restart is what has the change @@ -2757,14 +2807,15 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) } try { - runRequestedSessionRestarts(false); + runRequestedSessionRestarts(); } catch (Throwable restartFailure) { /* - * Nothing is left to try: the change is listed, uncommitted and unowned, so any - * later session restart of this domain delivers it again. This goes with the - * throwable which is rethrown below rather than being reported on its own. + * Nothing is left to try here: the change is listed, uncommitted and unowned, + * and the restart which threw has asked for one again, so the session restart + * the state checkpointer runs delivers it again. This goes with the throwable + * which is rethrown below rather than being reported on its own. */ suppress(recoveryFailure, restartFailure); } @@ -3760,74 +3811,58 @@ private boolean recoverFromReplayFailure( * run it: a restart which is already under way may have started before this change * was released, and the delivery it asked for would then have been turned down as a * duplicate of a change a replay thread still owned. + * + * A replay thread which is stopping - the number of them is being changed - asks for + * the restart all the same: nothing else would ask for the change it just released, + * and the ServerState would stay behind it for good. What it asks for is not owed the + * backoff, though: the backend is not what is going away. Neither is what the thread + * an OutOfMemoryError is ending asks for, for the same reason. The wait belongs to the + * request rather than to the thread which runs it, or a hand-back which is owed none + * would spend the wait another request is owed - and the other way around. */ - sessionRestartRequested.set(true); - /* - * A replay thread which is stopping - the number of them is being changed - restarts - * the session all the same: nothing else would ask for the change it just released, - * and the ServerState would stay behind it for good. It does not sit through the - * backoff on its way out, though: the backend is not what is going away. Neither does - * the thread an OutOfMemoryError is ending, for the same reason - and the restart is - * run rather than left to be asked for again, because that thread will not be there to - * run it, and a change nobody asks for again holds this domain's ServerState back. - */ - runRequestedSessionRestarts(!replayThreadShutdown.get() && !outOfMemory); + sessionRestarts.request(replayThreadShutdown.get() || outOfMemory + ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF); + runRequestedSessionRestarts(); return true; } /** * Restarts the session as long as changes which could not be replayed are waiting to be * delivered again. - * - * @param wait whether to leave the backend some time to recover between two restarts */ - private void runRequestedSessionRestarts(boolean wait) + private void runRequestedSessionRestarts() { /* * The outer loop is what makes a request which was made while this thread was giving * up the recovery its own: the thread which made it found the recovery taken and left * it to this one. */ - while (sessionRestartRequested.get() && replayFailureRecovery.compareAndSet(false, true)) + while (sessionRestarts.isPending() && replayFailureRecovery.compareAndSet(false, true)) { try { - while (sessionRestartRequested.getAndSet(false)) + for (SessionRestart restart = sessionRestarts.take(); + restart != SessionRestart.NONE; + restart = sessionRestarts.take()) { - boolean restarted = false; try { - restartSession(wait); - restarted = true; + restartSession(restart == SessionRestart.AFTER_BACKOFF); } - finally + catch (Throwable t) { - if (!restarted) - { - /* - * The request is put back where it was taken from. The flag is read and - * cleared before the restart runs, so a restart which ends abruptly - the - * session is stopped first, and starting it again creates a listener thread, - * which the operating system can refuse - would otherwise leave this domain - * with no session and with nothing left to ask for one. - * - * What a request left standing buys is bounded, and the bound is worth - * stating. Its two readers are the roads out of a failed and of an abandoned - * replay of this domain, and with no listener thread nothing is delivered - * anymore: the replays left to run are the changes already taken off the - * session - the ones waiting in the replay queue, and the ones parked as - * dependencies. One of those failing finds the request standing and runs the - * restart, which starts from a clean state, since disableService() drops the - * listener thread which was never started. Once they are spent, the domain - * stays down until it is disabled and enabled back, or the server is - * restarted. That is said where it can be heard: a refused thread is an - * OutOfMemoryError, and one which leaves recoverFromReplayFailure() or - * abandonReplay() ends the replay thread it is met on, so the uncaught - * exception handler of DirectoryThread writes the line and raises the alert, - * with the start of the listener thread in the trace. - */ - sessionRestartRequested.set(true); - } + /* + * The request is taken before the restart runs, so a restart which could not + * run - the session is stopped first, and starting it again creates a listener + * thread, which the operating system can refuse - has to ask for one again: + * the session has been stopped and was not started back, and nothing else + * would ask - no change is delivered over a session which is down, so no + * replay fails and no thread comes back here. The request is asked for with + * the backoff whatever it was made with, since a session which can not be + * started is the very thing that wait is for. + */ + sessionRestarts.giveBack(SessionRestart.AFTER_BACKOFF); + throw t; } } } @@ -3838,13 +3873,74 @@ private void runRequestedSessionRestarts(boolean wait) } } + /** + * Runs the session restart this domain was asked for and which no thread of its own + * ran. + *
+ * A restart is asked for by the thread which released the change it could not replay, + * and that thread usually runs it. It does not always: a replay thread on its way out + * hands its change back and leaves, and a restart which threw where it starts the + * session again asks for one anew rather than take the request away with it. Nothing + * would run what is left standing - a session which is down delivers no change, so no + * replay fails and no thread comes back for it - and this domain would sit out of the + * topology, with the changes it did not replay owned by the replication server and its + * ServerState stopped behind them. + *
+ * Not run while a total update is being processed, in either direction: a restart stops + * the session the total update runs over. An import into this replica reads its entries + * from that session and would end on the ones which had arrived - and {@code disabled} + * does not say an import is running, since {@code preBackendImport()} keeps the backend + * events this domain is the cause of from disabling it. An export from this replica + * publishes its entries over it, and {@code exportLDIFEntry()} gives the export up as + * {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT} once the broker has been stopped + * under it, which leaves the replica it was initializing to be initialized again. This + * thread is the one which can afford to wait: the request stays standing, and it comes + * back here once a second, so the restart is run as soon as the total update is over. + * The change the restart was asked for waits for as long as the total update takes, and + * the ServerState with it; the replay of this domain keeps running in the meantime. + */ + private void runPendingSessionRestart() + { + if (shutdown.get() || disabled || ieRunning() || !sessionRestarts.isPending()) + { + return; + } + try + { + runRequestedSessionRestarts(); + } + catch (Throwable t) + { + /* + * The restart which threw has asked for one again, so this thread runs it again + * once the backoff has been waited out. It must not end on it: nothing would save + * the ServerState of this domain anymore, and shutdown() waits for this thread. + * + * The report is guarded on its own, the way the report of a give-back which failed + * is: building the line walks the stack of the throwable, and on the road out of a + * JVM which has just refused an allocation that is a throw of its own, which would + * end this thread all the same. The restart is asked for again already, and the next + * run of it reports again if it fails again. + */ + try + { + logger.error(ERR_REPLAY_SESSION_RESTART_FAILED, + getBaseDN(), stackTraceToSingleLineString(t)); + } + catch (Throwable reportFailure) + { + // Nothing is left to say it with, and the report must not end this thread. + } + } + } + /** * Gives a change back to the replication server when this replay thread stops before it * could apply it. *
- * The change is not owned by anyone anymore and it was never applied, so the session is - * restarted for it to be delivered again: it is left out of the ServerState, and the - * changes which follow it are held back until it is replayed. + * The change is not owned by anyone anymore and it was never applied, so a session + * restart is asked for to have it delivered again: it is left out of the ServerState, + * and the changes which follow it are held back until it is replayed. * * @param csn the CSN of the change this thread was replaying */ @@ -3863,8 +3959,15 @@ private void abandonReplay(CSN csn) * is started back - one line per change would say otherwise. */ logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN()); - sessionRestartRequested.set(true); - runRequestedSessionRestarts(false); + /* + * Asked for rather than run here. The threads of the pool are stopped one after the + * other and joined, so every one of them which was replaying a change would stop and + * start the session on its way out, one restart per change abandoned and none of them + * waiting - while the configuration change which is stopping them waits for all of + * them. The state checkpointer runs one restart for the lot a moment later, which is + * all the replication server needs to send every change which was handed back. + */ + sessionRestarts.request(SessionRestart.NOW); } /** @@ -3874,6 +3977,7 @@ private void abandonReplay(CSN csn) private void restartSession(boolean wait) { final long stoppedSession; + final long wakes; synchronized (serviceStateLock) { if (sessionHasAnOwner()) @@ -3884,6 +3988,7 @@ private void restartSession(boolean wait) } disableService(); stoppedSession = getSessionGeneration(); + wakes = sessionRestartBackoffWakes(); } if (wait) { @@ -3893,7 +3998,7 @@ private void restartSession(boolean wait) * is not held under the lock, or a domain being disabled for an import would wait * it out. */ - waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet()); + waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet(), wakes); } synchronized (serviceStateLock) { @@ -3913,6 +4018,7 @@ private void restartSession(boolean wait) */ return; } + failSessionRestartIfATestAskedFor(); enableService(); } } @@ -3923,12 +4029,36 @@ private void restartSession(boolean wait) * fast as the replication server can send them. * * @param restarts how many times in a row the session was restarted already + * @param wakes how many times the backoff had been woken when the session was stopped, + * so that a wake given since ends the wait whether it found the wait under way + * or not */ - private void waitBeforeSessionRestart(int restarts) + private void waitBeforeSessionRestart(int restarts, long wakes) { + final long until = monotonicNowInMs() + + Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS); try { - Thread.sleep(Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS)); + /* + * Waited on a monitor rather than slept through, so that a domain which is going + * away or is being disabled is not waited out: the thread which holds this wait is + * a replay thread the shutdown of the pool joins, or the state checkpointer the + * shutdown of the domain waits for. The session it would start back is one the + * domain is stopping anyway. + * + * The wake is counted rather than only checked for by its flag: a domain which was + * disabled and enabled back has its flag cleared again, and a wait which read the + * flag only would go on for a session the restart has nothing left to start. + */ + synchronized (sessionRestartBackoff) + { + for (long left = until - monotonicNowInMs(); + left > 0 && !shutdown.get() && !disabled && wakes == sessionRestartBackoffWakes; + left = until - monotonicNowInMs()) + { + sessionRestartBackoff.wait(left); + } + } } catch (InterruptedException e) { @@ -3942,6 +4072,85 @@ private void waitBeforeSessionRestart(int restarts) } } + /** + * Has the next session restarts of this domain fail where they would start the session + * again. + *
+ * Only there for the tests: nothing asks a restart to fail in production, and what this + * stands in for - anything thrown out of {@link #enableService()}, which stops at no + * failure the callers of {@link ReplicationBroker#start()} report - can not be provoked + * from the outside. + * + * @param failures how many session restarts must fail before one is allowed to run + */ + @VisibleForTesting + public void failNextSessionRestarts(int failures) + { + sessionRestartFailuresToInject.set(failures); + } + + /** + * Returns how many of the session restart failures {@link #failNextSessionRestarts(int)} + * asked for have not been spent yet. + *
+ * Only there for the tests, which read it to tell a restart which failed from one which + * was never run. + * + * @return how many session restarts are still to fail + */ + @VisibleForTesting + public int getSessionRestartFailuresLeft() + { + return sessionRestartFailuresToInject.get(); + } + + /** + * Asks for a session restart the way a replay thread on its way out does, without + * running it. + *
+ * Only there for the tests, which have no other way to leave a request standing at a + * time of their choosing: the one a replay thread makes is made and run in one go, and + * the requests which stand across a span nothing runs them in - the domain disabled, or + * being imported into - are made in a window between a thread's read of the flag and the + * flag being set, which no test can hit on purpose. + */ + @VisibleForTesting + public void requestSessionRestart() + { + sessionRestarts.request(SessionRestart.NOW); + } + + /** + * Returns how many times in a row the session was restarted without a change being + * replayed in between: the count the backoff of the next restart is computed from. + *
+ * Only there for the tests, which read it to see a restart reach its backoff rather than + * wait a delay out and hope it began: the count is bumped on the way into the wait, so + * the restart of {@code n} restarts in a row is at its backoff, or a few instructions + * short of it, once this returns {@code n} - and a wake given in those instructions is + * counted, so the wait sees it all the same. + * + * @return how many session restarts in a row the domain has run + */ + @VisibleForTesting + public int getConsecutiveSessionRestarts() + { + return consecutiveSessionRestarts.get(); + } + + /** + * Fails this session restart when a test asked for it, and does nothing at all + * otherwise. + */ + private void failSessionRestartIfATestAskedFor() + { + if (sessionRestartFailuresToInject.getAndUpdate(left -> Math.max(left - 1, 0)) > 0) + { + throw new IllegalStateException("the session of domain " + getBaseDN() + + " could not be started again, as a test asked"); + } + } + /** * Generate a new CSN and insert it in the pending list. * @@ -4710,9 +4919,47 @@ public void disable() * is gone with the pending changes, so a leftover request would have a replay thread * stop and start the session once for a delivery which can not come. */ - sessionRestartRequested.set(false); + sessionRestarts.clear(); consecutiveSessionRestarts.set(0); } + // Woken outside the lock it does not hold: a restart which is waiting has nothing + // left to wait for, the session it would start being one this domain is not serving. + wakeSessionRestartBackoff(); + } + + /** + * Has a session restart which is sitting through its backoff stop waiting, and one which + * is about to sit through it not wait at all: the wake is counted (see + * {@link #sessionRestartBackoffWakes}). + */ + private void wakeSessionRestartBackoff() + { + synchronized (sessionRestartBackoff) + { + sessionRestartBackoffWakes++; + sessionRestartBackoff.notifyAll(); + } + } + + /** + * Returns how many times the backoff has been woken so far, for a session restart to + * take under the lock it stops the session under: the wake of a {@link #disable()} which + * has yet to take that lock is then one the restart sees, whether it is waiting by then + * or not, and one which took it already has left the restart nothing to stop - unless an + * {@link #enable()} took it since as well. The restart then stops the session + * {@code enable()} started and sits the whole backoff for it, as it is owed: its request + * was taken before either clear could reach it, and the wake it would have ended on is + * spent. It costs one restart of a session which delivers the change anyway, in a window + * between the take and the block a {@code disable()} and an {@code enable()} would both + * have to fit in. The wake of {@link #shutdown()} needs no counting - the flag it stands + * for never comes back. + */ + private long sessionRestartBackoffWakes() + { + synchronized (sessionRestartBackoff) + { + return sessionRestartBackoffWakes; + } } /** @@ -4830,6 +5077,16 @@ public void enable() { synchronized (serviceStateLock) { + /* + * Cleared here as well as by disable(): a request made in the window between a + * replay thread's read of the flag and disable()'s own clear - abandonReplay() reads + * it, then logs, then asks - survives the whole of the disabled span, and the state + * checkpointer would restart the session started below within the second for a + * change which is gone with the pending changes. Every request standing here is that + * one: the domain has been disabled since anything could ask, and the session started + * below asks for everything the ServerState loaded below does not cover. + */ + sessionRestarts.clear(); try { loadDataState(); @@ -5343,7 +5600,7 @@ protected void importBackend(InputStream input) throws DirectoryException * goes with them: the caller starts the session again from the reloaded state. */ remotePendingChanges.clear(); - sessionRestartRequested.set(false); + sessionRestarts.clear(); consecutiveSessionRestarts.set(0); importingData = false; } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java new file mode 100644 index 0000000000..a854b5aab4 --- /dev/null +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java @@ -0,0 +1,124 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import java.util.concurrent.atomic.AtomicReference; + +/** + * The session restart a replication domain has been asked for and has not run yet. + *
+ * A change which was released without being replayed is delivered again only over a new + * session, so the thread which released it asks this domain for one. The request is what + * the restart is run from, and it outlives the thread which made it: several threads + * releasing changes at once are answered by one restart, and a restart which could not + * run gives the request back rather than take it away with it - a change nobody asks for + * again holds the ServerState of this domain back for as long as the server is up. + *
+ * The wait a failing backend is owed belongs to the request rather than to the thread
+ * which runs it. A replay which failed asks for the restart with the wait, and a replay
+ * thread on its way out asks for it without - the backend is not what is going away - and
+ * either thread can end up running what the other asked for: the one which is owed no
+ * wait must not spend the wait the other's request was made with, and the one which is
+ * owed the wait must not have its request run without it, whichever of the two runs it.
+ */
+class SessionRestartRequests
+{
+ /**
+ * What a domain has been asked to do with its session, from what is not being asked for
+ * to what is asked for most insistently: {@link #merge(SessionRestart)} keeps the
+ * furthest down this list, so that a request is never answered by less than it asked
+ * for.
+ */
+ enum SessionRestart
+ {
+ /** Nothing is being asked for: every request made has been run. */
+ NONE,
+ /** The session is to be restarted as soon as a thread can run it. */
+ NOW,
+ /**
+ * The session is to be restarted once the backend has been left the time to recover
+ * which the restarts made in a row have climbed to.
+ */
+ AFTER_BACKOFF;
+ }
+
+ private final AtomicReference
+ * It is taken before the restart is run rather than once it has run: a change released
+ * while the restart was under way is not one that restart asks for - its delivery would
+ * have been turned down as a duplicate of a change a replay thread still owned - so the
+ * request it makes must outlive the restart which was already running.
+ *
+ * @return what is being asked for, {@link SessionRestart#NONE} when nothing is
+ */
+ SessionRestart take()
+ {
+ return requested.getAndSet(SessionRestart.NONE);
+ }
+
+ /**
+ * Asks again for a restart which was taken and could not be run.
+ *
+ * What is asked for again is not what {@link #take()} returned: the caller asks for the
+ * restart with the backoff whether or not the request it took was made with one, since
+ * a session which could not be started is the very thing that wait is for. A request
+ * made while the restart was running is not undone by it - the two are merged, and the
+ * one which asks for more wins.
+ *
+ * @param restart what the caller which could not run the restart asks for again
+ */
+ void giveBack(SessionRestart restart)
+ {
+ merge(restart);
+ }
+
+ /** Forgets what this domain was asked for, its pending changes being gone with it. */
+ void clear()
+ {
+ requested.set(SessionRestart.NONE);
+ }
+
+ /**
+ * Returns whether a restart is being asked for.
+ *
+ * @return {@code true} when a restart has been asked for and not run yet
+ */
+ boolean isPending()
+ {
+ return requested.get() != SessionRestart.NONE;
+ }
+
+ private void merge(SessionRestart restart)
+ {
+ requested.accumulateAndGet(restart,
+ (standing, asked) -> standing.compareTo(asked) >= 0 ? standing : asked);
+ }
+}
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index d28000e54a..c6b8fc1718 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -667,7 +667,8 @@ ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain
timeout instead
ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \
server after the replay which owned it was unwound: %s. The change has been released without its \
- failure being counted, and the session is being restarted so that the change is delivered again
+ failure being counted, and a restart of the session is asked for so that the change is delivered \
+ again
WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \
the replay of one of its changes to finish. A change which reaches the backend from now on \
is not recorded in the ServerState being saved, so the replication server sends it again \
@@ -679,6 +680,10 @@ WARN_REPLAY_DRAIN_INTERRUPTED_320=Domain "%s" is going down and was interrupted
WARN_CHANGELOG_READ_AGAIN_FOR_MISSING_CHANGES_321=Replication server RS(%d) holds changes for domain \
"%s" which never reached the message queue of %s: its state %s is behind the state %s of the \
domain although it is being served from that queue. Reading the changelog again to send them
+ERR_REPLAY_SESSION_RESTART_FAILED_325=Could not restart the replication session of domain \
+ "%s" for the changes which could not be replayed: %s. The session is left stopped and is \
+ started again a moment later; the changes it is being restarted for are not recorded as \
+ replayed and are still owned by the replication server
ERR_REPLICATION_DOMAIN_CONFIG_CHANGE_FAILED_326=Could not apply a configuration change to the \
replication domain on "%s": %s
NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327=The configuration change was applied to the \
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 77d0acfdd3..d57e0f99ca 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 @@ public class UpdateOperationTest extends ReplicationTestCase
*/
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;
@@ -2354,6 +2368,259 @@ public void call() throws Exception
}
}
+ /**
+ * 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
* The exporter is a broker of this test, so that the test says when the entries arrive: the
* change is replayed while the import is waiting for them - or, for the request, while the
@@ -284,6 +287,44 @@ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() throws Exceptio
}
}
+ /**
+ * A session restart which stood while the import ran was asked for by a replay thread
+ * for a change given back before the total update owned the session, and that change is
+ * forgotten with the pending changes when the imported data replaces the ServerState:
+ * the session started back at the end of the import asks for everything the imported
+ * state does not cover. Run, the request would stop that session once for a delivery
+ * which can not come. The request is made here by hand, in the place of one made
+ * between a replay thread's read of the owner and the import claiming the session.
+ *
+ * The restart is the state checkpointer's to run, within its first tick after the total
+ * update has released the session, so the pin is that the failure it would meet is never
+ * spent: a restart which ran would have spent it, and would have left the session it
+ * stopped down.
+ */
+ @Test(timeOut = 120_000)
+ public void aRequestWhichStoodWhileTheImportRanIsNotRunOnceItIsOver() throws Exception
+ {
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ domain.requestSessionRestart();
+ domain.failNextSessionRestarts(1);
+ try
+ {
+ finishImport(exported);
+
+ // Two ticks of the checkpointer: a request standing when the import ends is run on the first.
+ Thread.sleep(2000);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1, "the request which stood while"
+ + " the import ran was run against the session started back at its end");
+ assertTrue(domain.isConnected(), "the session started back at the end of the import"
+ + " was stopped for a request made before it");
+ }
+ finally
+ {
+ domain.failNextSessionRestarts(0);
+ }
+ }
+
/**
* Has the exporter start a total update into this replica, and returns once the backend
* of the domain is deregistered for it: from then on the import is reading the session,
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java
new file mode 100644
index 0000000000..632f0d9adb
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java
@@ -0,0 +1,380 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.plugin;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.forgerock.opendj.ldap.ModificationType.*;
+import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.protocols.internal.InternalClientConnection.*;
+import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
+import static org.testng.Assert.*;
+
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ResultCode;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.core.ModifyOperation;
+import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.OperationContext;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.types.Entry;
+import org.opends.server.types.OperationType;
+import org.testng.annotations.Test;
+
+/**
+ * Tests that a domain which is going down, or is being disabled, does not wait out the
+ * backoff of the session restart its state checkpointer is sitting through: the session
+ * that restart would start back is one the domain is stopping anyway, so the wait is on a
+ * monitor which {@code shutdown()} and {@code disable()} wake, rather than slept through.
+ * And that a request which stood while the domain was disabled is not run once it is
+ * enabled back: the change it was made for is gone with the pending changes.
+ *
+ * The checkpointer is the thread put through the wait here because it is the one whose
+ * wait can be seen from a test: {@code shutdown()} waits for it, and it is what saves the
+ * ServerState once the domain is enabled back. A replay thread sitting through the same
+ * wait is one thread of a shared pool, and its absence shows nowhere.
+ */
+@SuppressWarnings("javadoc")
+public class SessionRestartBackoffTest extends ReplicationTestCase
+{
+ private static final int RS_ID = 602;
+ private static final int DS_ID = 1;
+ private static final int BROKER_ID = 2;
+ private static final int GROUP_ID = 1;
+
+ /**
+ * How many session restarts in a row the checkpointer is left at the backoff of.
+ *
+ * The first two fail, and the third is the one this test acts on: it is owed a backoff
+ * of three seconds, which a wait that is not woken holds whoever is waiting for the
+ * checkpointer through most of. The test waits to see the restart reach that backoff
+ * rather than act a delay after the second failure and hope the backoff has begun by
+ * then: the checkpointer takes the request back on its next tick, up to a second after
+ * it reported that failure, and a tick pushed past a fixed delay would have the test act
+ * on a request which stands unwaited - which a wait that is slept through survives, so
+ * the run would prove nothing about the wake.
+ */
+ private static final int RESTARTS_IN_A_ROW = 3;
+
+ /**
+ * How long {@code shutdown()} or {@code disable()} may take when the backoff is woken:
+ * either drains a replay which is not running, stops a session which is already stopped,
+ * and saves the ServerState - {@code shutdown()} by waiting for the checkpointer to do it
+ * once more. A checkpointer which sleeps the backoff through while it holds the monitor
+ * the wake is given on holds either of them for the seconds the backoff has left instead.
+ */
+ private static final long WAKE_BOUND_IN_MS = 1500;
+
+ /**
+ * How long the checkpointer may take to save a change made once the domain is enabled
+ * back: its next tick, at most a second away, plus the save. One which was left asleep
+ * in the backoff {@code disable()} cut saves nothing until the backoff is out.
+ */
+ private static final long CHECKPOINT_BOUND_IN_MS = 2500;
+
+ /**
+ * How long a request left standing is given to be run by the checkpointer: two of its
+ * ticks, where one which is standing when the domain is enabled back is run on the first.
+ */
+ private static final long LEFTOVER_REQUEST_BOUND_IN_MS = 2000;
+
+ @Test
+ public void aDomainWhichIsShutDownDoesNotWaitOutTheBackoffOfItsSessionRestart()
+ throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ LDAPReplicationDomain domain = null;
+ ReplicationBroker broker = null;
+ try
+ {
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestShutdownDb");
+ domain = startDomain(baseDN, rsPort);
+ assertTrue(domain.isConnected(), "the domain did not connect to its replication server");
+ broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000);
+
+ leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.shutdown");
+
+ final long started = System.nanoTime();
+ domain.shutdown();
+ final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started);
+
+ assertTrue(tookMs < WAKE_BOUND_IN_MS, "shutdown() waited " + tookMs
+ + " ms: it sat out the backoff of the session restart the state checkpointer"
+ + " was waiting through, for a session the domain was stopping anyway");
+ }
+ finally
+ {
+ release(domain, broker, replicationServer);
+ }
+ }
+
+ @Test
+ public void aDomainWhichIsDisabledDoesNotWaitOutTheBackoffOfItsSessionRestart()
+ throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ LDAPReplicationDomain domain = null;
+ ReplicationBroker broker = null;
+ try
+ {
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestDisableDb");
+ domain = startDomain(baseDN, rsPort);
+ assertTrue(domain.isConnected(), "the domain did not connect to its replication server");
+ broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000);
+
+ final DN entryDN = leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.disable");
+
+ /*
+ * The data of this domain is about to be replaced, then has been. disable() is timed
+ * as shutdown() is: its wake takes the monitor the checkpointer waits on, so a
+ * checkpointer which sleeps the backoff through while holding that monitor holds
+ * disable() for the rest of it - and everything below would then start late enough
+ * for the save to land inside its bound all the same.
+ */
+ final long started = System.nanoTime();
+ domain.disable();
+ final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started);
+ assertTrue(tookMs < WAKE_BOUND_IN_MS, "disable() waited " + tookMs
+ + " ms: it sat out the backoff of the session restart the state checkpointer"
+ + " was waiting through, for a session disable() was cutting anyway");
+ domain.enable();
+ assertTrue(domain.isConnected(), "the domain did not come back up once enabled");
+
+ /*
+ * A change of this replica's own, for the checkpointer to save: nothing but the
+ * checkpointer writes the ServerState while the domain is up, so the change reaching
+ * the saved state says that the checkpointer is ticking rather than still asleep in
+ * the backoff disable() cut. It is read as the CSN the change was given, since the
+ * change the restart was asked for is delivered again over the new session and is
+ * recorded in that state as well.
+ */
+ final ModifyOperation modify = getRootConnection().processModify(
+ modifyRequest(baseDN, REPLACE, "description", "the checkpointer is ticking"));
+ assertEquals(modify.getResultCode(), ResultCode.SUCCESS,
+ modify.getAdditionalLogItems().toString());
+ final CSN csn = OperationContext.getCSN(modify);
+ assertNotNull(csn, "the change of this replica's own was given no CSN");
+
+ final long deadline = System.nanoTime() + MILLISECONDS.toNanos(CHECKPOINT_BOUND_IN_MS);
+ while (!persistedServerState(baseDN).cover(csn))
+ {
+ assertTrue(System.nanoTime() < deadline, "the state checkpointer did not save the"
+ + " ServerState within " + CHECKPOINT_BOUND_IN_MS + " ms of the domain being"
+ + " enabled back: it was left asleep in the backoff of the session restart"
+ + " disable() cut");
+ Thread.sleep(50);
+ }
+
+ // The change the restart was asked for is delivered over the session enable() started.
+ assertNull(getEntry(entryDN, 30000, false),
+ "the change was not delivered again over the session the domain was enabled with");
+ }
+ finally
+ {
+ release(domain, broker, replicationServer);
+ }
+ }
+
+ /**
+ * A request which stood while the domain was disabled was made for a change which is
+ * gone with the pending changes, and the session {@code enable()} starts asks for
+ * everything the ServerState it loads does not cover: run, the request would stop and
+ * start that session once for a delivery which can not come. The request is made here by
+ * hand, in the place of one made between a replay thread's read of the flag and the
+ * clear {@code disable()} does after it.
+ *
+ * The restart is the checkpointer's to run, within its first tick after the domain is
+ * enabled back, so the pin is that the failure it would meet is never spent: a restart
+ * which ran would have spent it, and would have left the session it stopped down.
+ */
+ @Test
+ public void aRequestWhichStoodWhileTheDomainWasDisabledIsNotRunOnceItIsEnabledBack()
+ throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ LDAPReplicationDomain domain = null;
+ try
+ {
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestEnableDb");
+ domain = startDomain(baseDN, rsPort);
+ assertTrue(domain.isConnected(), "the domain did not connect to its replication server");
+
+ domain.disable();
+ domain.requestSessionRestart();
+ domain.failNextSessionRestarts(1);
+ domain.enable();
+ assertTrue(domain.isConnected(), "the domain did not come back up once enabled");
+
+ Thread.sleep(LEFTOVER_REQUEST_BOUND_IN_MS);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1, "the request which stood while"
+ + " the domain was disabled was run against the session enable() started");
+ assertTrue(domain.isConnected(),
+ "the session enable() started was stopped for a request made before it");
+ }
+ finally
+ {
+ release(domain, null, replicationServer);
+ }
+ }
+
+ /**
+ * Leaves the state checkpointer of the provided domain sitting through the backoff of
+ * a session restart, with the session stopped.
+ *
+ * A change whose replay fails for longer than it is retried in place has the session
+ * restarted for it to be delivered again, and the first two restarts fail where they
+ * would start the session back: the one the replay thread runs, and the one the
+ * checkpointer runs for the request it gave back. The checkpointer runs the third, and
+ * it is the checkpointer whichever thread the second failure fell to, since the replay
+ * thread leaves on the failure it meets and no delivery brings one back over a session
+ * which is down. That third restart is owed the backoff of three restarts in a row, and
+ * this returns once the checkpointer has reached it, with the whole of it still to wait.
+ *
+ * @return the DN of the entry the change which is waiting to be delivered again deletes
+ */
+ private DN leaveTheCheckpointerInTheBackoff(
+ LDAPReplicationDomain domain, ReplicationBroker broker, String uid) throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: uid=" + uid + "," + domain.getBaseDN(),
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: " + uid,
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final String uuid = getEntry(entry.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ /*
+ * Unavailable for the attempts in place of the first delivery and two more, so that
+ * the delivery a session enabled back brings ends in the change being applied.
+ */
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+ domain.failNextSessionRestarts(2);
+
+ final CSNGenerator gen = new CSNGenerator(BROKER_ID, 0);
+ broker.publish(new DeleteMsg(entry.getName(), gen.newCSN(), uuid));
+
+ final long deadline = System.nanoTime() + SECONDS.toNanos(30);
+ while (domain.getSessionRestartFailuresLeft() > 0)
+ {
+ assertTrue(System.nanoTime() < deadline,
+ "the two session restarts which were asked to fail did not both run within 30 s");
+ Thread.sleep(50);
+ }
+ /*
+ * The third restart bumps the count on its way into the wait, so the checkpointer is
+ * at the backoff, or a few instructions short of it, once the count says three - and a
+ * wake given in those instructions is counted, so it ends the wait all the same.
+ */
+ while (domain.getConsecutiveSessionRestarts() < RESTARTS_IN_A_ROW)
+ {
+ assertTrue(System.nanoTime() < deadline, "the session restart of " + RESTARTS_IN_A_ROW
+ + " in a row did not reach its backoff within 30 s");
+ Thread.sleep(50);
+ }
+ assertFalse(domain.isConnected(),
+ "the session was started back before this test could act on the domain");
+ return entry.getName();
+ }
+
+ /**
+ * Lets go of what {@link #leaveTheCheckpointerInTheBackoff} set up, and of the domain and
+ * the replication server of the case - each step whatever the one before it threw, or a
+ * case which failed on its way up would leave its domain registered for the next one to
+ * replace silently.
+ */
+ private void release(LDAPReplicationDomain domain, ReplicationBroker broker,
+ ReplicationServer replicationServer) throws Exception
+ {
+ try
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ if (domain != null)
+ {
+ domain.failNextSessionRestarts(0);
+ }
+ if (broker != null)
+ {
+ broker.stop();
+ }
+ }
+ finally
+ {
+ try
+ {
+ if (domain != null)
+ {
+ MultimasterReplication.deleteDomain(domain.getBaseDN());
+ }
+ }
+ finally
+ {
+ remove(replicationServer);
+ }
+ }
+ }
+
+ private ServerState persistedServerState(DN baseDN) 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;
+ }
+
+ /**
+ * Creates and starts a domain on the provided base DN, and asserts nothing about it: the
+ * caller holds the domain before it looks at it, so that a case which fails there still
+ * has it to delete.
+ */
+ private LDAPReplicationDomain startDomain(DN baseDN, int rsPort) throws Exception
+ {
+ final SortedSet