From 41f5692c778b797fe09b5a658af37e4bc1da83ad Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 24 Sep 2026 11:56:08 +0000
Subject: [PATCH] [#1061] Ask for the session restart under an owner as well, and leave it to the state checkpointer (#1062)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 8
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 373 ++++++++++++++++++++++++++-
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 235 ++++++++++-------
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java | 74 +++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java | 42 +++
5 files changed, 608 insertions(+), 124 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index 678fc25..5126a65 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -2847,8 +2847,9 @@
* the state checkpointer runs one restart for every change the threads of the
* pool hand back on their way out, rather than each of them running one while
* the configuration change which is stopping them waits. A domain whose session
- * has an owner is left alone the way the give-back left it - it asked for nothing
- * there, which is why it handed back no change to run a restart for - and a
+ * has an owner is left alone the way the give-back left it - its request there
+ * stands for the state checkpointer to run once the owner is gone, which is why it
+ * handed back no change to run a restart for (see sessionHasAnOwner()) - and a
* request another thread left standing is not this one's to spend on a restart
* which is refused where it runs.
*/
@@ -3574,8 +3575,9 @@
// The ack has been published and the change is given back: the replication server
// delivers it again, now or - while a total update is being processed over the
// session - once it is over, when the restart which was held runs or, on the import
- // direction, when the session is started from the reloaded state. There is nothing
- // left to replay here.
+ // direction, when the session is started from the reloaded state, or the one the
+ // state checkpointer restarts once a total update which never began lets go of it.
+ // There is nothing left to replay here.
return;
}
@@ -4023,6 +4025,24 @@
*/
remotePendingChanges.replayFailed(csn);
+ /*
+ * This change is not owned by anyone anymore, so the session has to be restarted for
+ * the replication server to deliver it again. Ask for the restart before trying to
+ * 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.
+ */
+ sessionRestarts.request(replayThreadShutdown.get() || outOfMemory
+ ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF);
+
if (sessionHasAnOwner())
{
/*
@@ -4030,12 +4050,15 @@
* this thread's to restart. Restarting the one which is being stopped would leave a
* broker and a listener thread behind on a domain whose alert generator, flush
* thread and RSUpdater are already gone; restarting the one an import streams over
- * would end the import on the entries which had arrived. The change given back here
- * is forgotten with the rest of the pending changes when the ServerState is loaded
- * again, from the backend or from the imported data, and the session started then
- * asks for everything that state does not cover - or, when the total update it was
- * given back for never begins, it is asked for by the next restart
- * (see sessionHasAnOwner()).
+ * would end the import on the entries which had arrived. The request made above is
+ * left standing, the way it is on every road which releases a change under an owner
+ * (see sessionHasAnOwner()): the owners which forget the pending changes forget the
+ * request with them, and the one which does not - a total update which never begins
+ * - leaves it to the state checkpointer, which runs it once the owner is gone and
+ * has the replication server send the change again. The line which says the change
+ * is being asked for again is not logged here: a server which is shutting down
+ * abandons every change in flight, and none of them is asked for again before it is
+ * started back.
*/
return true;
}
@@ -4054,23 +4077,6 @@
*/
warned = logReplayRetryWarning(csn, failure);
}
- /*
- * This change is not owned by anyone anymore, so the session has to be restarted for
- * the replication server to deliver it again. Ask for the restart before trying to
- * 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.
- */
- sessionRestarts.request(replayThreadShutdown.get() || outOfMemory
- ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF);
if (!runRequestedSessionRestarts() && warned)
{
/*
@@ -4281,16 +4287,15 @@
* to be restarted, and a session which starts is given its receive window anew.
* <p>
* On a domain whose session has an owner - the domain itself, going away, or a total
- * update into it, from the moment it is asked for - they are released and nothing more,
- * the way {@code abandonReplay()} hands a change back on that road (see
- * {@link #sessionHasAnOwner()}): there is no session of this thread's to restart, and
- * the restart it would ask for is refused where it runs. The domain forgets its pending
- * changes on its way down, the import forgets them at its end, and a change released
- * for a total update which never begins stays listed until the next failed replay of
- * this domain restarts the session, which has the replication server send it again. A
- * line which says the replication server sends the change again would not hold on any
- * of these - a server which is shutting down abandons every change in flight, and none
- * of them is delivered again before it is started back.
+ * update into it, from the moment it is asked for - they are released and asked for, and
+ * nothing more, the way {@code abandonReplay()} hands a change back on that road (see
+ * {@link #sessionHasAnOwner()}): there is no session of this thread's to restart, so the
+ * request is left standing for the state checkpointer, which runs it once the owner is
+ * gone - or finds it forgotten, with the pending changes it was made for, by the domain
+ * on its way down or by the import at its end. A line which says the replication server
+ * sends the change again would not hold on every one of these - a server which is
+ * shutting down abandons every change in flight, and none of them is delivered again
+ * before it is started back - so none is logged, and the deliveries are not counted.
* <p>
* A replay thread which is stopping calls this through
* {@link #giveBackChangesParkedByStoppingThread()}, for every domain of this server: what
@@ -4307,7 +4312,7 @@
* restart asked for them - on a thread which is not stopping, and on a domain whose
* session has no owner - and reports what that restart did for them. Empty when
* this thread had parked none, and empty on a domain whose session has an owner,
- * where nothing is asked for. The list is the one the release allocated: nothing is
+ * where the request is left standing for the state checkpointer. The list is the one the release allocated: nothing is
* allocated for the answer on the road out of a JVM which has run out of memory
*/
private List<CSN> giveBackParkedChanges(SessionRestart restart)
@@ -4317,19 +4322,19 @@
{
return parked;
}
- if (sessionHasAnOwner())
- {
- // The domain owns its session, or a total update does: both forget the pending
- // changes, and neither leaves a session for this thread to restart. Nothing was asked
- // for here, so the caller has nothing of this road's to run or to report.
- return Collections.emptyList();
- }
/*
* Asked for before the changes are reported: a throw out of the report - the JVM which
* unwound this replay is out of memory - must not lose the restart which is what brings
* them back.
*/
sessionRestarts.request(restart);
+ if (sessionHasAnOwner())
+ {
+ // The domain owns its session, or a total update does: neither leaves a session for
+ // this thread to restart, and the request stands for the state checkpointer. The
+ // caller has nothing of this road's to run or to report.
+ return Collections.emptyList();
+ }
for (CSN csn : parked)
{
incProcessedUpdates();
@@ -4344,17 +4349,50 @@
* <p>
* 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.
+ * and the changes which follow it are held back until it is replayed. A change which
+ * this thread did not give back - it is not listed anymore, or another thread owns it -
+ * is asked for by nobody here.
*
* @param csn the CSN of the change this thread was replaying
*/
private void abandonReplay(CSN csn)
{
- remotePendingChanges.replayFailed(csn);
+ if (!remotePendingChanges.replayFailed(csn))
+ {
+ /*
+ * Not listed anymore: the owner which forgot the pending changes - disable() with
+ * enable(), or the import at its end - did it while this thread was on its way here,
+ * and it starts the session again from the ServerState it loaded, which asks for
+ * everything that state does not cover. A request made now would outlive the clear
+ * of that owner and have the state checkpointer stop and start the session it has
+ * just started, for a change which is gone. The other road to false - the change is
+ * owned by another thread, which took it over since - needs no restart either: that
+ * thread is the one which replays it or gives it back.
+ */
+ return;
+ }
+ /*
+ * 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.
+ *
+ * Asked for whether or not the session has an owner, the way every road which releases
+ * a change asks (see sessionHasAnOwner()): the domain on its way down and the import at
+ * its end forget the request with the pending changes, and a total update which never
+ * begins leaves it to the state checkpointer. Asked for once the change is released and
+ * not before: asked for first, it could be taken and run by another thread while this
+ * one still owned the change, and the delivery the new session brought would be turned
+ * down as the duplicate of a change a replay thread owns.
+ */
+ sessionRestarts.request(SessionRestart.NOW);
if (sessionHasAnOwner())
{
- // The domain, or the import into it, owns its session, and the pending changes are
- // forgotten with the ServerState on its way down or at the end of the import.
+ // The domain, or the import into it, owns its session: a server which is shutting
+ // down abandons every change in flight, and none of them is delivered again before
+ // it is started back, so the line below would not hold.
return;
}
/*
@@ -4363,15 +4401,6 @@
* is started back - one line per change would say otherwise.
*/
logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN());
- /*
- * 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);
}
/**
@@ -4517,10 +4546,10 @@
* running it.
* <p>
* 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.
+ * time of their choosing without a change released for it: the one a replay thread makes
+ * on a domain whose session has no owner is made and run in one go, and one made under
+ * an owner stands for a change which is listed, and which the owner forgets or the
+ * checkpointer has delivered again.
*/
@VisibleForTesting
public void requestSessionRestart()
@@ -5492,24 +5521,25 @@
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.
+ * Cleared here as well as by disable(): a request made after disable()'s own clear -
+ * every road which releases a change asks for the restart, whether or not the domain
+ * owns its session, and a replay which outlasted the drain disable() waits for
+ * releases its change under the flag - 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.
*
- * The deliveries folded into no warning are forgotten here for the same window: a
- * thread which is recording a failed replay reads the flag, then folds the delivery
- * - recoverFromReplayFailure() logs before it asks - then asks, and runs what it
- * asked for itself, which restartSession() turns down on a domain which owns its
- * session. So on that road it is the fold rather than the request which outlives
- * disable()'s clear, and the first warning over the data loaded back would count a
- * delivery of a change which went with the pending changes. What reads the count is
- * a warning, and none is logged between disable() and here short of that thread's
- * own, so a test tells this zeroing and disable()'s apart by nothing: they stand or
- * fall together.
+ * The deliveries folded into no warning are forgotten here for a window of the same
+ * kind: a thread which is recording a failed replay asks, then reads the flag, then
+ * folds the delivery and runs what it asked for, which restartSession() turns down
+ * on a domain which owns its session. A read which found the domain enabled a moment
+ * before disable() set the flag leaves a fold which outlives disable()'s clear, and
+ * the first warning over the data loaded back would count a delivery of a change
+ * which went with the pending changes. What reads the count is a warning, and none
+ * is logged between disable() and here short of that thread's own, so a test tells
+ * this zeroing and disable()'s apart by nothing: they stand or fall together.
*/
sessionRestarts.clear();
foldedReplayRetryWarnings.set(0);
@@ -5532,12 +5562,11 @@
* stopping one: enableService() ends with startListenService(), so the listener it
* starts can list a delivery and hand it to a replay thread while this method is
* still running. A replay thread which reads a flag that still says "disabled"
- * gives the change up at the top of its replay loop, and abandonReplay() does not
- * ask for it again - a domain on its way down owns its session - so the change is
- * left listed, uncommitted and owned by nobody. Nothing would replay it: the
- * replication server only sends it again over a session which is restarted, so this
- * domain's ServerState, and every change which depends on that one, would be held
- * back for as long as the session lives.
+ * gives the change up at the top of its replay loop, and abandonReplay() releases it
+ * with a request standing - made after the clear above, so nothing clears it - which
+ * the state checkpointer runs within its tick: the change would be delivered again,
+ * but over a restart of the session just started rather than to the replay it had
+ * been handed to.
*/
disabled = false;
boolean started = false;
@@ -6021,10 +6050,12 @@
* uncommitted change - the state would never move past it, and the replication
* server does not send a change again which the state it is given covers. Nothing
* listed a change meanwhile: the listener thread is the one running this import,
- * and the replay threads gave up every attempt while the flag was set. The restart
- * a replay thread may have asked for before the total update owned the session
- * goes with them, as do the deliveries folded into no warning: the caller starts
- * the session again from the reloaded state.
+ * and the replay threads gave up every attempt while the flag was set. The restarts
+ * they asked for on their way out, while the total update owned the session, and
+ * any asked for before it did, go with them, as do the deliveries folded into no
+ * warning: the caller starts the session again from the reloaded state. A replay
+ * thread which reaches its give-up after this clear finds its change unlisted and
+ * asks for nothing.
*/
remotePendingChanges.clear();
sessionRestarts.clear();
@@ -6324,13 +6355,25 @@
* arrives over that session, so a restart made while it is on its way loses it, and the
* import which follows reads its entries over the same session - stopping it ends the
* import on the entries which had arrived. The import starts the next session itself,
- * from the state it loaded. A change given back while the total update owned the session
- * is not asked for again by anyone until then; if no import follows - the request was
- * refused, or gave up waiting - it stays listed until the next failed replay restarts
- * the session, which has the replication server send it again with everything after it.
- * Listed, it holds the ServerState back as well: a commit moves the state no further than
- * the oldest uncommitted change, so the state in memory, and the one persisted from it,
- * stop at the change until that restart.
+ * from the state it loaded.
+ * <p>
+ * A change released while the session has an owner is asked for again all the same - every
+ * road which releases one asks, owner or not - and the request is what is left to the
+ * owner: none of the roads runs the restart it asks for under an owner, since
+ * {@link #restartSession(boolean)} refuses it and the request would be spent on the
+ * refusal, and the state checkpointer holds every request for as long as the session has
+ * an owner. Two owners forget the pending changes and the request with them, when the
+ * ServerState they went with is replaced: {@link #disable()} and {@link #enable()}, and
+ * the import at its end - the session started then asks for everything the state it
+ * loaded does not cover. The third does not: a total update which is asked for and never
+ * begins - the request refused, or given up as unanswered - replaces nothing, and the
+ * request left standing under it is run by the checkpointer within its tick of the owner
+ * letting go, which has the replication server send the change again with everything
+ * after it. Left listed and asked for by nobody, the change would hold the ServerState
+ * back for good on a domain which then goes quiet: a commit moves the state no further
+ * than the oldest uncommitted change, so the state in memory, and the one persisted from
+ * it, would stop at the change until the next failed replay of this domain restarted
+ * the session (issue #1061).
* <p>
* A configuration change is refused while a total update runs - by the listener of the
* domain entry and by the one of its external changelog entry - so what reaches the
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
index 5835b12..085d147 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -332,8 +332,12 @@
* now is the double replay the ownership is there to prevent (issue #922).
*
* @param csn the CSN of the change whose replay failed
+ * @return whether the change was given back: {@code false} when it is not listed as an
+ * uncommitted change anymore - the domain forgot its pending changes while it was
+ * being replayed - or when another thread owns it, in which case nothing was
+ * released here and there is no delivery for a session restart to ask for
*/
- public void replayFailed(CSN csn)
+ public boolean replayFailed(CSN csn)
{
pendingChangesWriteLock.lock();
try
@@ -343,7 +347,9 @@
{
change.setOwner(null);
changeBeingReplayed.remove(Thread.currentThread(), csn);
+ return true;
}
+ return false;
}
finally
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java
new file mode 100644
index 0000000..1c84d7c
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java
@@ -0,0 +1,74 @@
+/*
+ * 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.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.zip.DataFormatException;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.core.ModifyOperation;
+import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.types.LDAPException;
+import org.opends.server.types.Modification;
+
+/**
+ * A ModifyMsg whose replay waits, before anything is attempted, until the test lets it go.
+ * <p>
+ * The operation is built first thing by a replay, before any lock is taken and before the
+ * change is checked against the ones it may depend on, so a replay which waits here holds
+ * the change as one being replayed - listed, uncommitted and owned by its thread - for as
+ * long as the test wants, while every other replay of the domain runs: a change which
+ * follows this one on the same entry is parked as waiting for it, and nothing has failed,
+ * so no session restart has been asked for. Let go, the replay runs to its end and commits
+ * the change as any other.
+ * <p>
+ * Such a message can not travel the protocol: it is handed to the domain rather than
+ * published, and replayed on a thread of the test.
+ */
+final class ModifyMsgWhoseOperationWaitsToBeBuilt extends ModifyMsg
+{
+ private final CountDownLatch letGo;
+
+ /**
+ * @param letGo the latch the replay waits on before the operation is built
+ */
+ ModifyMsgWhoseOperationWaitsToBeBuilt(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID, CountDownLatch letGo)
+ {
+ super(csn, dn, mods, entryUUID);
+ this.letGo = letGo;
+ }
+
+ @Override
+ public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
+ throws LDAPException, IOException, DataFormatException
+ {
+ try
+ {
+ letGo.await();
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new IOException("interrupted while waiting to be let go", e);
+ }
+ return super.createOperation(connection, newDN);
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
index f99cca8..6c1ed1c 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -120,7 +120,7 @@
assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
"a change a replay thread owns must not be taken over (OPENDJ-1115)");
- pendingChanges.replayFailed(csn);
+ assertTrue(pendingChanges.replayFailed(csn), "the change its owner gave back was not released");
assertEquals(pendingChanges.getQueueSize(), 1, "the change must stay listed as pending");
assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed");
@@ -499,15 +499,18 @@
assertTrue(pendingChanges.putRemoteUpdate(delivery));
assertTrue(pendingChanges.markInProgress(delivery));
+ final AtomicReference<Boolean> releasedByTheOtherThread = new AtomicReference<>();
runAndJoin(new Runnable()
{
@Override
public void run()
{
- pendingChanges.replayFailed(csn);
+ releasedByTheOtherThread.set(pendingChanges.replayFailed(csn));
}
});
+ assertEquals(releasedByTheOtherThread.get(), Boolean.FALSE,
+ "a release which was ignored must say so: nothing was given back to ask for again");
assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
"a change another thread is replaying must not be taken over");
assertEquals(pendingChanges.getQueueSize(), 1);
@@ -515,7 +518,8 @@
"a release which was ignored must leave the change with the thread which owns it");
// The thread which owns the change is still the one which decides its fate.
- pendingChanges.replayFailed(csn);
+ assertTrue(pendingChanges.replayFailed(csn), "the change its owner gave back was not released");
+ assertFalse(pendingChanges.replayFailed(csn), "a change given back once was released again");
assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
"the change its owner gave back must be taken over by the next delivery");
/*
@@ -529,6 +533,38 @@
}
/**
+ * A release of a change which is not listed as an uncommitted one anymore gives nothing
+ * back, and says so: a replay thread which reaches its give-up after the pending changes
+ * were forgotten - by the domain disabled and enabled again, or by the import at its end -
+ * must not ask for a session restart for a change the ServerState loaded then accounts for
+ * (issue #1061).
+ */
+ @Test
+ public void replayFailedReleasesNothingOnceTheChangeIsForgottenOrCommitted() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+
+ final CSN forgotten = generator.newCSN();
+ final DeleteMsg forgottenDelivery = deleteMsg(forgotten, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(forgottenDelivery));
+ assertTrue(pendingChanges.markInProgress(forgottenDelivery));
+ pendingChanges.clear();
+ assertFalse(pendingChanges.replayFailed(forgotten),
+ "a change the pending changes forgot was reported as given back");
+
+ final CSN committed = generator.newCSN();
+ final DeleteMsg committedDelivery = deleteMsg(committed, "uuid-2");
+ assertTrue(pendingChanges.putRemoteUpdate(committedDelivery));
+ assertTrue(pendingChanges.markInProgress(committedDelivery));
+ pendingChanges.commit(committed);
+ assertFalse(pendingChanges.replayFailed(committed),
+ "a change which was committed was reported as given back");
+ assertTrue(state.cover(committed));
+ }
+
+ /**
* A change which was parked because it depends on another one is handed out by
* {@code getNextUpdate()} to whichever replay thread calls it first once the change it
* was waiting for is gone, rather than replayed by the thread which parked it as a
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
index e921beb..0919df6 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -72,15 +72,20 @@
* one thing it must not do is stop the session the import is reading (issue #956). The same
* holds from the moment the total update is asked for: the answer to the request arrives
* over that session, so a replay which fails while it is on its way must not restart it.
- * A restart asked for before the total update took the session, and left standing for the
- * length of it, is not run once it is over either: the change it was asked for is gone with
- * the ServerState the import replaced. The changes a replay which is unwound had parked as
- * waiting for another one are released on the same terms, and nothing more is done for them
- * (issue #954).
+ * A restart asked for while the total update owns the session is left standing for the
+ * length of it, and is not run once it is over: the change it was asked for is gone with the
+ * ServerState the import replaced. The changes a replay which is unwound had parked as
+ * waiting for another one are released on the same terms (issue #954). A total update which
+ * is asked for and never begins - the request is refused, or gives up waiting for its answer
+ * - replaces nothing, and the request left standing under it is what has the changes
+ * released under it delivered again (issue #1061).
* <p>
* 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
- * exporter is holding the answer.
+ * exporter is holding the answer, which it may never give. A change which has to be delivered
+ * again is published through the replication server, which is what has it to send again;
+ * the replay queue of the domain is the test's, so every delivery is replayed when, and on
+ * the thread, the test says.
* <p>
* The claim of a total update this replica did not ask for is made by the listener thread
* under no lock, so a restart of the session which reads no owner a moment before that claim
@@ -309,12 +314,13 @@
/**
* The changes a replay which is unwound had parked as waiting for another change are
- * released and nothing more while a total update owns the session (issue #954): no
- * session restart is asked for them - the one it would ask for is refused where it runs,
- * and the request would be spent on it - and they are neither reported as changes the
- * replication server sends again, which it does not before the import has replaced the
- * data, nor counted as processed. That is the road a change a stopping replay thread
- * abandons takes on this domain, and the give-back of the parked changes takes it too.
+ * released while a total update owns the session, and the session is left to the owner
+ * (issue #954): the restart asked for them is not run - it is refused where it runs, and
+ * the request would be spent on it - and they are neither reported as changes the
+ * replication server sends again, which it does not before the total update has let go
+ * of the session, nor counted as processed. That is the road a change a stopping replay
+ * thread abandons takes on this domain, and the give-back of the parked changes takes it
+ * too.
* <p>
* Pinned on the import road because it is the one road with an owner which a test holds
* open for as long as it needs: the request is on its way until the exporter answers it,
@@ -329,7 +335,7 @@
* back, and the error which ends a replay thread is caught here instead.
*/
@Test(timeOut = 120_000)
- public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain() throws Exception
+ public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayLeavesTheSessionToTheOwner() throws Exception
{
final Entry entry = TestCaseUtils.addEntry(
"dn: cn=renamedSince," + EXAMPLE_DN,
@@ -399,10 +405,10 @@
"the change parked by the replay which was unwound must be given back");
assertEquals(getMonitorAttrValue(baseDN, "replayed-updates"), processed,
"a change released while a total update owns the session must not be counted as"
- + " processed: no session sends it again before the import has replaced the data");
+ + " processed: no session sends it again before the total update lets go of it");
assertThat(errorLogRecordsOf(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK.ordinal(), parked))
.as("the change was reported as one the replication server sends again, which it does"
- + " not before the import has replaced the data")
+ + " not before the total update lets go of the session")
.isEmpty();
assertTrue(domain.isConnected(), "the session the answer to the request arrives over was stopped");
@@ -417,13 +423,287 @@
}
/**
+ * A change released while a total update which never begins owns the session is asked for
+ * again under the owner, and delivered again over the session the state checkpointer
+ * restarts once the owner is gone (issue #1061).
+ * <p>
+ * A total update this replica asked for owns the session from the request on, and a change
+ * whose replay fails meanwhile is released and left to the owner: the domain forgets its
+ * pending changes on its way down, and the import forgets them at its end - but a request
+ * which is refused, or which gives up waiting for its answer, replaces nothing and forgets
+ * nothing. Released and asked for by nobody, the change would stay listed and uncommitted
+ * until the next failed replay of this domain restarted the session, and the ServerState -
+ * which a commit moves no further than the oldest uncommitted change - would stop at it
+ * with everything behind it. So the restart is asked for under the owner as well, and the
+ * state checkpointer, which holds every request for as long as the total update owns the
+ * session, runs it within its tick of the owner letting go.
+ * <p>
+ * The road pinned here is the one a change whose attempts in place are spent takes: every
+ * attempt ends on an entryUUID search which does not run. The request gives up through
+ * the watchdog of the initialize task, which is the one road out of an unanswered request
+ * a test can take at a time of its choosing.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeReleasedUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() 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());
+
+ // The request is out, and the exporter never answers it.
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ final LDAPUpdateMsg delivered = publishAndAwaitDelivery(new ModifyMsg(csn,
+ DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "released while the request was on its way"), entryUUID));
+ // A restart run on this thread would be back up before any read of the session: this counts it.
+ domain.failNextSessionRestarts(1);
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replay(delivered);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+ assertFalse(domain.getServerState().cover(csn),
+ "the change whose replay fails must stay listed as one which is not in the data");
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was warned about as one the replication server sends again, which it"
+ + " does not while the total update owns the session")
+ .isEmpty();
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the replay whose attempts were spent");
+ domain.failNextSessionRestarts(0);
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(csn, 30_000, "the change released under the"
+ + " request was not delivered again once the request gave up: nothing asked for the"
+ + " session restart which has the replication server send it again");
+ replay(again);
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("description"))
+ .as("the change delivered again was not applied").isNotEmpty();
+ assertTrue(domain.getServerState().cover(csn),
+ "the change delivered again was applied and not recorded: it is still listed");
+ }
+
+ /**
+ * A change a stopping replay thread abandons while a total update which never begins owns
+ * the session takes the same road (issue #1061): abandoned at the top of its first attempt
+ * without being counted against its budget, released, asked for again under the owner,
+ * and delivered again once the owner is gone.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeAbandonedUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() 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());
+
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ final LDAPUpdateMsg delivered = publishAndAwaitDelivery(new ModifyMsg(csn, entry.getName(),
+ generatemods("description", "abandoned while the request was on its way"), entryUUID));
+ // The thread of this test is one which is stopping: the change is abandoned unapplied.
+ assertTrue(domain.markInProgress(delivered), "the delivery was not handed to this thread");
+ domain.failNextSessionRestarts(1);
+ domain.replay(delivered, new AtomicBoolean(true));
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("description"))
+ .as("a change abandoned by a stopping thread was applied").isEmpty();
+ assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), csn))
+ .as("the change was reported as one the replication server sends again, which it"
+ + " does not while the total update owns the session")
+ .isEmpty();
+ assertEquals(getMonitorAttrValue(baseDN, "changes-with-failed-replay"), 0,
+ "the abandoned change was counted against its budget: it was never attempted");
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the thread which abandoned the change");
+ domain.failNextSessionRestarts(0);
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(csn, 30_000, "the change abandoned under the"
+ + " request was not delivered again once the request gave up: nothing asked for the"
+ + " session restart which has the replication server send it again");
+ replay(again);
+ assertTrue(domain.getServerState().cover(csn),
+ "the change delivered again was applied and not recorded: it is still listed");
+ }
+
+ /**
+ * A change a stopping replay thread abandons once the import has forgotten it asks for no
+ * session restart (issue #1061).
+ * <p>
+ * The import forgets the pending changes at its end, and the session it starts asks for
+ * everything the ServerState it loaded does not cover. A thread which read the flag while
+ * the import ran and reaches the give-back only after that finds its change unlisted: a
+ * restart asked for then would outlive the clear, and the state checkpointer would stop
+ * and start the session the import has just started, for a change which is gone.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeAbandonedOnceTheImportForgotItAsksForNoRestart() throws Exception
+ {
+ final String[] exported = exportedEntries();
+
+ // Owned by this thread before the import begins, and still owned when the import ends.
+ final CSN csn = gen.newCSN();
+ domain.processUpdate(new ModifyMsg(csn, DN.valueOf(IMPORTED_ENTRY_DN),
+ generatemods("description", "abandoned once the import forgot it"), IMPORTED_ENTRY_UUID));
+ final LDAPUpdateMsg delivered = queue.take().getUpdateMessage();
+ assertTrue(domain.markInProgress(delivered), "the delivery was not handed to this thread");
+
+ startImportInto(exported.length);
+ finishImport(exported);
+
+ domain.failNextSessionRestarts(1);
+ try
+ {
+ domain.replay(delivered, new AtomicBoolean(true));
+
+ // Two ticks of the checkpointer: a request standing now is run on the first.
+ Thread.sleep(2000);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1, "a change the import forgot"
+ + " asked for a restart of the session the import started at its end");
+ assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), csn))
+ .as("a change the import forgot was reported as one the replication server sends again")
+ .isEmpty();
+ }
+ finally
+ {
+ domain.failNextSessionRestarts(0);
+ }
+ }
+
+ /**
+ * A change the give-back released while a total update which never begins owns the session
+ * is asked for again under the owner by the give-back itself, and the request is left
+ * standing rather than spent (issue #1061).
+ * <p>
+ * The change it waited for is one another thread is replaying, held before its operation
+ * is built: nothing has failed, so no road but the give-back has asked for anything, and
+ * the request found standing once the owner is gone is the give-back's own. No restart is
+ * run on the thread of the give-back under the owner: one run there would be refused where
+ * it runs, and the case counts the restarts which get that far.
+ * <p>
+ * The replay which is unwound is this thread's, as in the case above: its change is
+ * applied, and the ack of its delivery runs out of memory.
+ */
+ @Test(timeOut = 120_000)
+ public void aParkedChangeGivenBackUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() 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 other = TestCaseUtils.addEntry(
+ "dn: cn=unwound," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: unwound",
+ "sn: unwound");
+ final String otherUUID = getEntryUUID(other.getName());
+
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ /*
+ * The change the parked one waits for: replayed by a thread of the test which is held
+ * before the operation is built, so the change is being replayed - listed, uncommitted,
+ * owned - for as long as the latch holds, and nothing has failed.
+ */
+ final CountDownLatch letGo = new CountDownLatch(1);
+ final CSN held = gen.newCSN();
+ domain.processUpdate(new ModifyMsgWhoseOperationWaitsToBeBuilt(held, entry.getName(),
+ generatemods("description", "the change being replayed by another thread"), entryUUID,
+ letGo));
+ final LDAPUpdateMsg heldMsg = queue.take().getUpdateMessage();
+ final Thread otherThread = new Thread(() ->
+ {
+ assertTrue(domain.markInProgress(heldMsg), "the held change must be the one listed");
+ domain.replay(heldMsg, SHUTDOWN);
+ }, "ReplayDuringImportTest replay held before its operation is built");
+ otherThread.start();
+ try
+ {
+ // Parked as waiting for the held change by this thread, which owns it from here on.
+ final CSN parked = gen.newCSN();
+ final LDAPUpdateMsg parkedDelivery = publishAndAwaitDelivery(new ModifyMsg(parked,
+ entry.getName(),
+ generatemods("description", "the change which was parked as a dependency"), entryUUID));
+ replay(parkedDelivery);
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
+ "a change which waits for one being replayed by another thread must be parked");
+
+ // The replay which is unwound while this thread still holds the parked change.
+ final CSN unwound = gen.newCSN();
+ domain.failNextSessionRestarts(1);
+ try
+ {
+ replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, other.getName(),
+ generatemods("description", "the replay of this change is unwound once it is applied"),
+ otherUUID));
+ Assert.fail("the replay was not unwound: the ack of the delivery must run out of memory");
+ }
+ catch (OutOfMemoryError unwinding)
+ {
+ // The error is the fixture's own, and this is the thread it would have ended.
+ }
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "the change parked by the replay which was unwound must be given back");
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the replay which was unwound");
+ domain.failNextSessionRestarts(0);
+
+ // The held change runs to its end: nothing fails, nothing asks for a restart.
+ letGo.countDown();
+ otherThread.join(30_000);
+ assertFalse(otherThread.isAlive(), "the held replay did not end once let go");
+ assertTrue(domain.getServerState().cover(held), "the held change was not recorded");
+ assertFalse(domain.getServerState().cover(parked),
+ "the parked change must stay listed as one which is not in the data");
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(parked, 30_000, "the parked change given back"
+ + " under the request was not delivered again once the request gave up: the session"
+ + " restart the give-back asked for under the owner was not run, or was spent");
+ replay(again);
+ assertTrue(domain.getServerState().cover(parked),
+ "the parked change delivered again was applied and not recorded: it is still listed");
+ }
+ finally
+ {
+ letGo.countDown();
+ otherThread.join(30_000);
+ }
+ }
+
+ /**
* 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.
+ * for a change it gave back - before the total update owned the session, or under the
+ * owner - 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 the one a failed replay makes.
* <p>
* 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
@@ -983,8 +1263,53 @@
private void replayMsg(UpdateMsg updateMsg) throws InterruptedException
{
domain.processUpdate(updateMsg);
- final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage();
- domain.markInProgress(ldapUpdate);
- domain.replay(ldapUpdate, SHUTDOWN);
+ replay(queue.take().getUpdateMessage());
+ }
+
+ /** Replays a delivery on the thread of this test, as a replay thread would. */
+ private void replay(LDAPUpdateMsg delivery)
+ {
+ assertTrue(domain.markInProgress(delivery), "the delivery is not the one listed: " + delivery);
+ domain.replay(delivery, SHUTDOWN);
+ }
+
+ /**
+ * Publishes a change through the replication server, which is what has it to deliver again
+ * once the session is restarted for it, and waits for the delivery to this replica.
+ */
+ private LDAPUpdateMsg publishAndAwaitDelivery(LDAPUpdateMsg msg) throws Exception
+ {
+ exporter.publish(msg);
+ return awaitDelivery(msg.getCSN(), 30_000, "the change published was not delivered");
+ }
+
+ /**
+ * Waits for the replication server to deliver the change to this replica: the listener
+ * thread of the domain puts it in the replay queue of the test, which takes it out.
+ */
+ private LDAPUpdateMsg awaitDelivery(CSN csn, long timeoutMs, String orElse) throws Exception
+ {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ while (queue.peek() == null)
+ {
+ assertTrue(System.currentTimeMillis() < deadline, orElse + " within " + timeoutMs + " ms");
+ Thread.sleep(50);
+ }
+ final LDAPUpdateMsg msg = queue.take().getUpdateMessage();
+ assertEquals(msg.getCSN(), csn, "another change than the one awaited was delivered");
+ return msg;
+ }
+
+ /**
+ * Has the total update this replica asked for give up on its request, the way the
+ * watchdog of the initialize task does once the request has waited two minutes for an
+ * answer: the total update never begins, and its context is released with nothing
+ * replaced, so the session has no owner anymore.
+ */
+ private void giveUpTheRequest()
+ {
+ assertTrue(domain.abortStalledInitializeFromRemote(0),
+ "the request was not the one waiting for an answer");
+ assertFalse(domain.ieRunning(), "the total update was given up and is still being processed");
}
}
--
Gitblit v1.10.0