From 80481f756d71bd58b4bda627758dcd774e9d5dcd Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 18 Sep 2026 14:45:54 +0000
Subject: [PATCH] [#986] Give back the changes a replay thread the pool stopped had parked (#988)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 112 +++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java | 15
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationRefusesAControl.java | 67 +
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 3
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ParkedChangeGiveBackTest.java | 373 ++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java | 87 ++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java | 72 ++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 114 +++
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 544 +++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 217 +++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ForEachDomainTest.java | 184 +++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied.java | 82 ++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java | 150 ++++
13 files changed, 1,968 insertions(+), 52 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 5e75be0..f68df24 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
@@ -2760,20 +2760,35 @@
* failing and is eventually given up on: handing it back bare would have this domain
* ask for it, and restart its session for it, for as long as the server is up.
*
- * The changes this thread parked as waiting for another change are left alone: they
- * are handed to whichever thread clears the change they are waiting for, and that
- * thread takes them over.
+ * The changes this thread parked as waiting for another change are given back too,
+ * and before the road above runs: that road restarts the session, and a change which
+ * is still owned when the replication server sends it again over it is turned down as
+ * a duplicate - the one delivery which could have taken it over (issue #954).
*
- * Which change this thread owns is read before anything is done with it, and that
- * read takes no lock and allocates nothing: everything below is gated on the answer,
- * so a lookup which threw in its turn - on the road out of a JVM which has just
- * refused an allocation - would leave the change listed, uncommitted and owned by a
- * thread which is about to end, which is the state this whole issue is about.
+ * Which change this thread owns is read first of all, and that read takes no lock and
+ * allocates nothing: everything below is gated on the answer, so a lookup which threw
+ * in its turn - on the road out of a JVM which has just refused an allocation - would
+ * leave the change listed, uncommitted and owned by a thread which is about to end,
+ * which is the state this whole issue is about. It is read before the parked changes
+ * are given back rather than after, because that give-back allocates and can throw on
+ * the same road, and the last resort below can only hand back a change it was told
+ * about.
*/
CSN owned = null;
try
{
owned = remotePendingChanges.getChangeOwnedByCurrentThread();
+ /*
+ * Asked for without the backoff on the two roads recoverFromReplayFailure() asks
+ * for it without on: a thread which is stopping, and one which an OutOfMemoryError
+ * is ending - the backend is not what is going away. What is asked for is not what
+ * is run: a request is never answered by less than it asked for, so a restart
+ * another road asked for with the backoff, or one given back with it, keeps its
+ * wait whichever thread runs it.
+ */
+ final boolean parkedGivenBack = giveBackParkedChanges(
+ replayThreadShutdown.get() || t instanceof OutOfMemoryError
+ ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF);
if (owned != null)
{
if (replayThreadShutdown.get() || shutdown.get() || disabled)
@@ -2808,6 +2823,27 @@
recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError);
}
}
+ if (parkedGivenBack && !replayThreadShutdown.get() && !sessionHasAnOwner())
+ {
+ /*
+ * The road the change this thread was replaying took may have run the restart the
+ * give-back asked for - they ask for the same one - and it may have had none to
+ * run: this thread owned no change, or the change it owned was given up on. What
+ * is still requested is run here rather than left standing: the state
+ * checkpointer would run it within its tick, so this is the latency of the
+ * delivery the changes which were handed back wait for, and nothing more - a
+ * request this thread leaves is not lost.
+ *
+ * A thread which is stopping leaves it standing, the way abandonReplay() does:
+ * 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: nothing was asked
+ * for on that road, and a request another thread left standing is not this
+ * one's to spend on a restart which is refused where it runs.
+ */
+ runRequestedSessionRestarts();
+ }
}
catch (Throwable recoveryFailure)
{
@@ -2819,7 +2855,15 @@
* 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. A
* restart which can not run here leaves its request standing, and the state
- * checkpointer of this domain runs it.
+ * checkpointer of this domain runs it: one which threw asks for itself again on
+ * its way out, and one a domain whose session has an owner would refuse is not
+ * run at all rather than spent on the refusal.
+ *
+ * The restart is asked for once the change is released and not before, the way
+ * every road which releases one asks: 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, with nothing left standing to ask for it again.
*/
if (owned != null)
{
@@ -2840,20 +2884,42 @@
{
suppress(recoveryFailure, reportFailure);
}
- try
+ }
+ try
+ {
+ /*
+ * Outside the guard above: two roads reach here with a request standing and no
+ * change of this thread's to hand back, and both are the parked changes' - the
+ * give-back which released them asks for the restart before it reports them, and
+ * a throw out of the report - the JVM which unwound this replay is out of memory
+ * - leaves the request standing; and a restart the parked road ran and which
+ * threw has asked for one again on its way out. The changes it released are
+ * listed, uncommitted and unowned, so the request is what brings them back, and
+ * this thread is the one there to run it (issue #954). A give-back which threw
+ * before it released anything left the parked changes as they were, owned by this
+ * thread and handed out by getNextUpdate() to whichever thread clears what they
+ * wait for: nothing here can do better for those.
+ *
+ * Not run on a domain whose session has an owner, the way no road of a failed
+ * replay runs it there: the restart is refused where it runs and the request
+ * would be spent on the refusal, while a request left standing is run by the
+ * state checkpointer once the owner is gone - or forgotten with the pending
+ * changes it was made for, when the owner forgets them on its way out.
+ */
+ if (!sessionHasAnOwner())
{
runRequestedSessionRestarts();
}
- catch (Throwable restartFailure)
- {
- /*
- * 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);
- }
+ }
+ catch (Throwable restartFailure)
+ {
+ /*
+ * Nothing is left to try here: the changes are listed, uncommitted and unowned,
+ * and the restart which threw has asked for one again, so the session restart
+ * the state checkpointer runs delivers them again. This goes with the throwable
+ * which is rethrown below rather than being reported on its own.
+ */
+ suppress(recoveryFailure, restartFailure);
}
// The error which unwound the replay is the one reported, whatever the give-back
// ran into on top of it.
@@ -3387,12 +3453,14 @@
* an OutOfMemoryError of its own, still owns its change: it is given back
* counted, and the thread ends on this error rather than on the one it stepped
* over. A replay which committed owns nothing anymore - commit() cleared the
- * owner, and the index the give-back reads, in the same step - so the give-back
- * is a no-op, and rightly so: a change which is in the data is not one to ask
- * for again. What that road steps over is getNextUpdate() below, so the changes
- * parked behind the committed change wait for the next replay of this domain to
- * hand them out. That is the trade #923 asks for: a thread which met this error
- * is not to carry on, not even for them.
+ * owner, and the index the give-back reads, in the same step - so the change it
+ * was replaying is not given back, and rightly so: a change which is in the data
+ * is not one to ask for again. What that road steps over is getNextUpdate()
+ * below, which hands out the changes parked behind the committed change: the
+ * ones this thread parked are given back on the way out of replay() and the
+ * session is restarted for them (issue #954), the ones other threads parked wait
+ * for the next replay of this domain to hand them out. That is the trade #923
+ * asks for: a thread which met this error is not to carry on, not even for them.
*/
throw e;
}
@@ -3968,6 +4036,25 @@
}
/**
+ * Gives back the changes a replay thread which is stopping parked in this domain.
+ * <p>
+ * Called by that thread on its way out (issue #986). The restart which brings them back
+ * is asked for and left standing, the way the thread leaves the request it makes for the
+ * change it abandons: the threads of the pool are stopped one after the other and joined,
+ * and each running a restart on its way out would have the configuration change which is
+ * stopping them wait for one restart per thread. The state checkpointer of this domain
+ * runs one restart for the lot within its tick, and holds it while a total update runs
+ * over the session, in either direction - a change delivered again before the pool which
+ * replaces this thread is up waits in the replay queue for it. Asked for without the
+ * backoff: what went away is a replay thread, not the backend, and these changes were
+ * never applied here.
+ */
+ void giveBackChangesParkedByStoppingThread()
+ {
+ giveBackParkedChanges(SessionRestart.NOW);
+ }
+
+ /**
* Restarts the session as long as changes which could not be replayed are waiting to be
* delivered again.
*/
@@ -4076,6 +4163,84 @@
}
/**
+ * Gives back the changes this replay thread parked as waiting for another change, on the
+ * way out of a replay which was unwound.
+ * <p>
+ * A parked change is handed out again by {@code getNextUpdate()} alone, which every
+ * replay loop of this domain runs once it is done with a change: a parked change is
+ * replayed by whichever thread clears the change it was waiting for. A thread whose
+ * replay was unwound is not on that road anymore - it takes the next delivery off the
+ * replay queue - so a change it parked would be left owned by a thread which is not
+ * coming back to it, while every redelivery of it is refused as a duplicate. On a domain
+ * which then goes quiet that change is where this replica's ServerState, and every change
+ * behind it from every master, stops (issue #954).
+ * <p>
+ * They are handed back without a failure being counted against them: they were never
+ * applied here, so the give-up budget which decides when this replica skips a change it
+ * can not apply is not this delivery's to spend, the way it is not for a change abandoned
+ * by a replay thread which is stopping.
+ * <p>
+ * The delivery which carried one published no ack - the ack of a parked change is
+ * published by the delivery which replays it - so it is counted as processed here, the
+ * way a delivery which is dropped rather than replayed is: that count is of the
+ * deliveries this replica took off the session, and these are over. The window they hold
+ * is not given back either, and does not need to be: the session they came over is about
+ * 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.
+ * <p>
+ * A replay thread which is stopping calls this through
+ * {@link #giveBackChangesParkedByStoppingThread()}, for every domain of this server: what
+ * it parked would be left owned by a thread which does not exist anymore, and every
+ * redelivery of a change a replay thread owns is refused as a duplicate (issue #986). The
+ * request it makes here is left standing for the state checkpointer, the way that thread
+ * leaves the request it makes for the change it abandons.
+ *
+ * @param restart what the session restart is asked for as: with the backoff a failing
+ * backend is owed, or without it on a thread which is stopping or which an
+ * OutOfMemoryError is ending
+ * @return whether any change was handed back: a change which nobody owns is one only a
+ * new delivery brings back, so the caller runs the restart asked for them - on
+ * a thread which is not stopping, and on a domain whose session has no owner
+ */
+ private boolean giveBackParkedChanges(SessionRestart restart)
+ {
+ final List<CSN> parked = remotePendingChanges.releaseParkedChangesOwnedByCurrentThread();
+ if (parked.isEmpty())
+ {
+ return false;
+ }
+ 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.
+ return true;
+ }
+ /*
+ * 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);
+ for (CSN csn : parked)
+ {
+ incProcessedUpdates();
+ logger.info(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK, csn, getBaseDN());
+ }
+ return true;
+ }
+
+ /**
* Gives a change back to the replication server when this replay thread stops before it
* could apply it.
* <p>
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
index 549d35a..3e1f4a9 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
@@ -33,6 +33,7 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -797,6 +798,77 @@
}
/**
+ * Runs an action on every replication domain of this server, each of them getting its
+ * turn whatever one of them threw.
+ * <p>
+ * The replay threads are shared by every domain of this server, so what concerns the
+ * pool is done over the domains rather than over the one a replay was last for: a thread
+ * which is stopping gives back what it parked in any of them (issue #986). A throw at one
+ * domain must not leave the ones after it as they were - with changes owned by a thread
+ * which does not exist anymore - so the first failure is thrown once the loop is over,
+ * the others suppressed under it where it records suppression: the error a JVM out of
+ * memory prepared beforehand does not - it was made without its constructor, so it keeps
+ * no list to record them in - and the JVM hands that one out as often as it is asked for
+ * one, so two domains can throw the same instance, and a throwable can not suppress
+ * itself. Recording a failure under the first allocates the list it goes in, so on the
+ * road this loop is for it can be refused in its turn: a failure which can not be
+ * recorded is dropped, and the domains after it still get their turn.
+ * <p>
+ * The iterator over the domains is the one allocation made before the first of them gets
+ * its turn: refused, on the way out of a thread an OutOfMemoryError is ending, it leaves
+ * every domain as it was, and it has no cheaper form. The action itself is not one: a
+ * caller on that road passes an instance it holds rather than one it makes there.
+ * <p>
+ * Not synchronized, and it must not become so: it is called by a replay thread on its way
+ * out, while {@link #stopReplayThreads()} holds the monitor of this class and waits for
+ * that thread to end.
+ *
+ * @param action what is done on each domain; it declares no checked exception, so what
+ * it throws is an Error or a RuntimeException, and that is what is thrown here
+ */
+ static void forEachDomain(Consumer<LDAPReplicationDomain> action)
+ {
+ Throwable failure = null;
+ for (LDAPReplicationDomain domain : domains.values())
+ {
+ try
+ {
+ action.accept(domain);
+ }
+ catch (Throwable domainFailure)
+ {
+ if (failure == null)
+ {
+ failure = domainFailure;
+ }
+ else if (failure != domainFailure)
+ {
+ try
+ {
+ failure.addSuppressed(domainFailure);
+ }
+ catch (OutOfMemoryError recordRefused)
+ {
+ /*
+ * The list the record goes in could not be allocated: the failure is dropped
+ * rather than allowed to end the loop, since the first one is what is reported
+ * and the domains after this one are still to be visited.
+ */
+ }
+ }
+ }
+ }
+ if (failure instanceof Error)
+ {
+ throw (Error) failure;
+ }
+ if (failure instanceof RuntimeException)
+ {
+ throw (RuntimeException) failure;
+ }
+ }
+
+ /**
* Gets the number of handled domain objects.
* @return The number of handled domain objects
*/
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 ec3ef7b..62629a0 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
@@ -17,7 +17,11 @@
*/
package org.opends.server.replication.plugin;
+import static java.util.Collections.*;
+
+import java.util.ArrayList;
import java.util.Iterator;
+import java.util.List;
import java.util.NoSuchElementException;
import java.util.SortedMap;
import java.util.SortedSet;
@@ -90,8 +94,10 @@
* this issue is about (issue #922).
* <p>
* A thread is entered here when it takes a change over and removed when it gives it back,
- * applies it, or parks it as waiting for another change - the parked ones are handed to
- * whichever thread clears what they wait for, so they are not this one's to give back.
+ * applies it, or parks it as waiting for another change - a parked change is not the one
+ * this thread is replaying, and giving it back is
+ * {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are
+ * waiting rather than this index (issue #954).
* <p>
* The entry of a thread is written by that thread and by nobody else, and that - not the
* lock - is what keeps the writes apart: the park in {@link #addDependency(PendingChange)}
@@ -545,9 +551,10 @@
* Returns the CSN of the change the calling thread is replaying, when it still owns one.
* <p>
* A thread owns the change it is replaying and the ones it parked as waiting for another
- * change. The parked ones are left out: they are handed to whichever thread clears the
- * change they are waiting for, and that thread takes them over, so giving one back here
- * would have the same change handed to two threads (issue #922).
+ * change. The parked ones are left out: they are not the change this thread is replaying,
+ * and giving one back is more than dropping its owner - it has to be unparked in the same
+ * step, or it would be handed out by two roads at once, which is what
+ * {@link #releaseParkedChangesOwnedByCurrentThread()} does (issues #922 and #954).
* <p>
* It is a plain read of {@link #changeBeingReplayed}: no lock is taken and nothing is
* allocated. This is what the give-back on the way out of an unwound replay asks first,
@@ -570,6 +577,95 @@
}
/**
+ * Gives back the changes the calling thread parked as waiting for another change, and
+ * takes them out of the changes which are waiting in the same step.
+ * <p>
+ * A parked change stays owned by the thread which parked it while that thread goes on
+ * to the changes which follow: {@link #getNextUpdate()} is what hands it out again, to
+ * whichever replay thread clears the change it was waiting for, and that thread takes it
+ * over. A replay which is unwound leaves the thread which parked it without that road -
+ * it takes the next delivery off the replay queue instead - so the change would be left
+ * owned by a thread which is never coming back to it, and every redelivery of a change a
+ * replay thread owns is refused as a duplicate (issue #954).
+ * <p>
+ * Unparking a change and giving it back is one step, under both locks, so that only one
+ * road can hand it out: a change which was released while it is still listed as waiting
+ * would be handed to the thread {@link #getNextUpdate()} gives it to and to the thread
+ * which takes over the delivery which follows - the double replay the ownership is there
+ * to prevent (OPENDJ-1115).
+ * <p>
+ * The changes stay listed and uncommitted, and stay among the changes the newer ones are
+ * checked against, the way a change whose replay failed does: they are not in the data,
+ * so they hold this domain's ServerState back and the changes which follow them keep
+ * waiting for them.
+ * <p>
+ * The changes another thread parked are left alone: a change is given back by the thread
+ * which owns it and by nobody else (issue #922). That thread may be inside the dependency
+ * checks which parked it - they park a change once per dependency it has - so a change
+ * released under it would be listed as waiting again a moment later, and handed out while
+ * the delivery which took it over is being replayed.
+ *
+ * @return the CSNs of the changes it gave back, oldest first; empty when this thread owns
+ * no parked change - the changes a thread parked stay its own, whichever replay
+ * parked them, until {@link #getNextUpdate()} hands them to the thread which
+ * cleared what they wait for or they are given back here
+ */
+ List<CSN> releaseParkedChangesOwnedByCurrentThread()
+ {
+ final Thread current = Thread.currentThread();
+ /*
+ * The second lock is taken inside the try of the first: taking a lock which is held
+ * by another thread allocates the node this one waits on, and this runs on the road
+ * out of a JVM which has just refused an allocation. A throw out of the second lock
+ * would otherwise unwind past the first with that one held by a thread which is
+ * ending, and every road which lists, commits or gives back a change would wait for
+ * it for good.
+ */
+ pendingChangesWriteLock.lock();
+ try
+ {
+ dependentChangesLock.lock();
+ try
+ {
+ if (dependentChanges.isEmpty())
+ {
+ // Nothing is waiting, which is the state every replay but a handful leaves behind.
+ return emptyList();
+ }
+ /*
+ * Sized for every change which is waiting, so that the one allocation of this
+ * method past the locks is made before anything is taken out of the set. The rule
+ * getNextUpdate() states for itself holds here: an allocation which fails once a
+ * change has been unparked and released - and this runs on the road out of a JVM
+ * which has just refused one - would have moved that change out of the hands which
+ * hand it out again, with the caller never told that it did.
+ */
+ final List<CSN> released = new ArrayList<>(dependentChanges.size());
+ final Iterator<PendingChange> it = dependentChanges.iterator();
+ while (it.hasNext())
+ {
+ final PendingChange change = it.next();
+ if (change.isOwnedBy(current))
+ {
+ it.remove();
+ change.setOwner(null);
+ released.add(change.getCSN());
+ }
+ }
+ return released;
+ }
+ finally
+ {
+ dependentChangesLock.unlock();
+ }
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
* Get the first update in the list that have some dependencies cleared.
* <p>
* The change is handed to the calling thread, which owns it from then on: it is
@@ -672,8 +768,10 @@
* parked one is handed to the thread which clears what it waits for, and one which is
* not listed here anymore is gone with the pending changes of a domain which was
* disabled. The owner stays as it is - it is what has getNextUpdate() hand the change
- * over rather than leave it to nobody - and the give-back on the way out of an
- * unwound replay leaves it alone (issue #922).
+ * over rather than leave it to nobody - and the give-back of the change a replay was
+ * unwound on leaves it alone (issue #922). What hands a parked change back is
+ * releaseParkedChangesOwnedByCurrentThread(), which unparks it in the same step so
+ * that the two roads can not hand it out at once (issue #954).
*/
changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
index 0dddf3a..27cffe6 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
@@ -24,6 +24,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
import org.opends.server.api.DirectoryThread;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -39,6 +40,15 @@
public class ReplayThread extends DirectoryThread
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /**
+ * The give-back a thread runs on every domain of this server on its way out, held here
+ * rather than written where it is run: a method reference is linked, and its instance
+ * made, where it is first run, and this one is first run on the way out of a thread -
+ * which an OutOfMemoryError may be ending, on the road this give-back is there for. Made
+ * when this class is loaded instead, on a thread which can allocate (issue #986).
+ */
+ private static final Consumer<LDAPReplicationDomain> GIVE_BACK_PARKED_CHANGES =
+ LDAPReplicationDomain::giveBackChangesParkedByStoppingThread;
private final BlockingQueue<UpdateToReplay> updateToReplayQueue;
private final ReentrantLock switchQueueLock;
@@ -77,6 +87,53 @@
logger.trace("Replication Replay thread starting.");
}
+ try
+ {
+ replayUntilStopped();
+ }
+ finally
+ {
+ /*
+ * The changes this thread parked as waiting for another change are handed out again
+ * by getNextUpdate() alone, which every replay loop of a domain runs once it is done
+ * with a change: a parked change is replayed by whichever thread clears the change it
+ * was waiting for. A thread which is stopping is not on that road anymore, so what it
+ * parked would be left owned by a thread which does not exist, while every redelivery
+ * of a change a replay thread owns is refused as a duplicate: on a domain which then
+ * goes quiet that change is where the ServerState of this replica, and every change
+ * behind it from every master, stops (issue #986).
+ *
+ * Given back by the thread which owns them, so that the rule every road which reads
+ * ownership follows holds on this one as well: a change is given back by the thread it
+ * was handed to and by nobody else (issue #922). It is also the one place which sees
+ * them all - the pool is shared by every domain of this server, while a replay knows
+ * only the domain it was replaying for.
+ *
+ * The session which brings them back is asked for and left standing, in every domain
+ * which got something back, and the state checkpointer of each of them runs it within
+ * its tick: a thread on its way out is not held for a session - the threads of the
+ * pool are stopped one after the other and joined, and each running a restart of its
+ * own would have the configuration change which is stopping them wait for one restart
+ * per thread - and a change delivered again before the pool which replaces this one
+ * is up waits in the replay queue for it. A thread which an OutOfMemoryError is ending
+ * gives back here what it parked in the domains it was not replaying for, on the same
+ * terms; the change it was replaying, and what it had parked in that same domain, were
+ * given back and asked for again on its way out of replay().
+ */
+ giveBackParkedChanges();
+ }
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("Replication Replay thread stopping.");
+ }
+ }
+
+ /**
+ * Takes the deliveries of the domains of this server off the shared replay queue and
+ * replays them, until this thread is stopped.
+ */
+ private void replayUntilStopped()
+ {
while (!shutdown.get())
{
try
@@ -145,9 +202,31 @@
logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(t));
}
}
- if (logger.isTraceEnabled())
- {
- logger.trace("Replication Replay thread stopping.");
- }
+ }
+
+ /**
+ * Gives back the changes this thread parked as waiting for another change, in every
+ * domain of this server.
+ * <p>
+ * A change which is given back stays listed and uncommitted, the way a change whose replay
+ * failed does: it is not in the data, so it holds the ServerState of its domain back and
+ * the changes which follow it keep waiting for it, until the delivery which takes it over
+ * replays it.
+ * <p>
+ * Every domain gets its turn whatever one of them threw: what can throw here is an
+ * allocation, on the way out of a thread an OutOfMemoryError may be ending - the iterator
+ * over the domains, before any of them is reached, then for each of them the list of what
+ * it released, made before anything is released, or the report of a change once it is,
+ * and between two domains the list a second failure is recorded in under the first, which
+ * the loop guards on its own - and the domains which follow would otherwise be left with
+ * changes owned by a thread which does not exist anymore, the state this give-back is
+ * for. A domain which threw past the release has asked for its restart already: the
+ * request is made before the report. The first failure is thrown once the loop is over,
+ * so that the uncaught exception handler of {@link DirectoryThread} writes the line and
+ * raises the alert.
+ */
+ private void giveBackParkedChanges()
+ {
+ MultimasterReplication.forEachDomain(GIVE_BACK_PARKED_CHANGES);
}
}
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
index a854b5a..51f13b3 100644
--- 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
@@ -58,6 +58,21 @@
private final AtomicReference<SessionRestart> requested =
new AtomicReference<>(SessionRestart.NONE);
+ SessionRestartRequests()
+ {
+ /*
+ * Every request is made on a replay-failure road, where an allocation may be what has
+ * just failed, and the first execution of merge() in a JVM allocates: the call site
+ * of its lambda and the VarHandle site inside accumulateAndGet() are linked when they
+ * are first run, and nothing runs them before a replay fails. Run once here, on a
+ * thread which can allocate, so that a request made on the road out of an
+ * OutOfMemoryError asks for nothing the JVM has just refused (issue #954). The same
+ * goes for what takes a request, which is run on the same roads.
+ */
+ merge(SessionRestart.NONE);
+ take();
+ }
+
/**
* Asks this domain to restart its session.
*
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 65f1ba9..f81040f 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -670,6 +670,9 @@
server after the replay which owned it was unwound: %s. The change has been released without its \
failure being counted, and a restart of the session is asked for so that the change is delivered \
again
+NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK_318=Change %s in domain "%s" was waiting for another change \
+ to be replayed when the replay thread which parked it went away. The change has not been recorded \
+ as replayed and is given back to the replication server, which still owns it and sends it 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 \
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 123ace3..325212b 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
@@ -95,6 +95,7 @@
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
import org.opends.server.types.RawModification;
+import org.opends.server.util.StaticUtils;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
import org.opends.server.util.TimeThread;
@@ -3583,6 +3584,508 @@
}
}
+ /**
+ * Test case for [Issue 954]: a change parked as waiting for another one is given back
+ * when the replay which parked it is unwound, and the session is restarted for it.
+ * <p>
+ * A change which waits for another one is parked and stays owned by the replay thread
+ * which parked it, while that thread goes on to the changes which follow: it is handed
+ * out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
+ * it is done, so it is replayed by whichever thread clears the change it was waiting for.
+ * A replay which is unwound leaves the thread which parked it without that road - it
+ * takes the next delivery off the shared queue instead, and never comes back to the
+ * change it parked - and every redelivery of a change a replay thread owns is refused as
+ * a duplicate. On a domain which then goes quiet that change is where this replica's
+ * ServerState, and every change behind it from every master, stops.
+ * <p>
+ * The replay which is unwound here is one which applied its change: the change it was
+ * replaying is committed and owns nothing anymore by the time the give-back on the way
+ * out of {@code replay()} runs, so the restart that give-back asks for is the one thing
+ * which has the parked change delivered again - a replay which failed would have asked
+ * for the same restart on the road of its own change. The parked change travels the
+ * replication server, and nothing but a new session brings it back.
+ */
+ @Test
+ public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain"));
+
+ final DN waitedOn = addEntryForChange("user.954.1");
+ final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
+ final DN other = addEntryForChange("user.954.2");
+ final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ domain.resetUnreplayedChangeAlertThrottle();
+ final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "no change of this domain is waiting for another one when this test starts");
+
+ // The replica the changes come from: one of its own, so that the CSNs of this test are
+ // never covered by the ServerState another one left behind.
+ final int serverId = 30;
+ final CSNGenerator gen = new CSNGenerator(serverId, TimeThread.getTime());
+ final CSN failing = gen.newCSN();
+ final CSN parked = gen.newCSN();
+ final CSN unwound = gen.newCSN();
+ final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
+ final String parkedDescription = "the change which was parked as a dependency";
+ final List<Modification> parkedMods = generatemods("description", parkedDescription);
+ final List<Modification> unwoundMods =
+ generatemods("description", "the replay of this change is unwound once it is applied");
+
+ /*
+ * The change whose replay fails is the barrier the parked change waits behind, so its
+ * budget must not be spent while this test is setting up: it is shortened once the
+ * change which was parked is back, and put back in the finally below.
+ */
+ setReplayGiveUpDelay("unlimited");
+ try
+ {
+ /*
+ * One replay thread, so that the change which is parked and the replay which is
+ * unwound after it are the same thread's: a parked change is left owned by the thread
+ * which parked it, and this is about a thread which does not come back to it.
+ */
+ setNumUpdateReplayThreads(1);
+ try
+ {
+ /*
+ * The change which is parked is published to the replication server rather than
+ * handed to the domain: a change which travelled a session is one the replication
+ * server sends again over the next session of this domain, and over nothing else -
+ * which is what the restart the give-back asks for has to be pinned against.
+ */
+ final ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ /*
+ * A change whose replay failed stays listed and uncommitted - it is what holds this
+ * domain's ServerState back - and stays among the changes the newer ones are checked
+ * against, so a change which follows it on the same entry has to wait for it. It has
+ * to be a change which is asked for again rather than stepped over: one whose
+ * operation is built and then refused, since #928 has a modify whose entry DN does
+ * not parse reported once and recorded as replayed.
+ */
+ deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
+ () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
+ "the change whose replay fails must be listed as one which is not in the data");
+
+ // The change which is parked as waiting for it by the replay thread it was given to.
+ broker.publish(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
+ assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
+ "a change which waits for one that is not in the data must be parked");
+
+ /*
+ * How many restarts in a row the session has been through, the barrier's among
+ * them: the restart the give-back asks for on the road out of a JVM which has run
+ * out of memory is owed no backoff, and a restart which waits the backoff out is
+ * the one road which moves this count. It can not move otherwise between here and
+ * the reading below: a replay which made it puts the count back to zero only once
+ * nothing is failing anymore, and the barrier keeps failing until it is given up.
+ */
+ final int restarts = domain.getConsecutiveSessionRestarts();
+
+ /*
+ * The replay which is unwound while that same thread still holds the parked change.
+ * It applies its change, and the ack of its delivery is where the JVM runs out of
+ * memory: that is the one throw from the ack which is not caught, past every catch
+ * the replay itself has, and it leaves replay() with the change this thread was
+ * replaying committed and owned by nobody. What the give-back on the way out finds
+ * to give back is the parked change alone, and the restart it asks for is the one
+ * road which has that change delivered again: the road a failed replay takes to
+ * ask for its own change is not run. An Error met replaying a change does not get
+ * here, and neither does any other throw from the ack: both are reported and the
+ * replay carries on to the road the change itself decided (issue #922).
+ *
+ * It is made on another entry, so that it is replayed rather than parked in its
+ * turn, and it is delivered until the parked change has been given back: a delivery
+ * of a change which is being replayed, or was applied, is refused as the duplicate
+ * it is.
+ */
+ deliverUntilMonitorReaches(domain, "dependent-changes-size", 0,
+ () -> new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, other, unwoundMods, otherUUID),
+ "a change parked by a replay which was unwound must be given back");
+
+ /*
+ * The thread which met the error is gone - an OutOfMemoryError ends the replay
+ * thread which met it (issue #923) - and it was the whole pool, so the delivery the
+ * restarted session brings waits in the replay queue: the pool is brought back for
+ * it, at any number now that the thread it had to be the same as is over. The change
+ * is parked again by the thread which takes it, the barrier being still there: that
+ * it is parked at all is what says a new session delivered it.
+ */
+ setNumUpdateReplayThreads(2);
+ assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
+ "the change which was given back must be delivered again by the session which"
+ + " was restarted for it");
+ assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
+ "a thread an OutOfMemoryError is ending must not wait the backoff out before"
+ + " it restarts the session for the changes it gave back");
+
+ /*
+ * The barrier is lifted, which lets the ServerState past it and hands the parked
+ * change to the thread which cleared it.
+ */
+ giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+ checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
+ "the change which was parked must be applied by the delivery which took it over");
+ }
+ finally
+ {
+ broker.stop();
+ }
+ assertEquals(resetNumUpdateReplayThreads(), 0,
+ "the number of replay threads could not be put back");
+ }
+ finally
+ {
+ /*
+ * Best-effort on the way out of a red, with no assertion to replace the failure
+ * which is being reported: on the normal road the number is put back, and checked,
+ * above.
+ */
+ resetNumUpdateReplayThreads();
+ }
+ }
+ finally
+ {
+ try
+ {
+ /*
+ * A red before the barrier was given up would leave it listed for the rest of this
+ * class: it never travelled the replication server, so no session restart brings a
+ * delivery which could give it up, and a change which is failing keeps every later
+ * session restart at its backoff and the ServerState behind it. The failure which
+ * is being reported is the one that matters, so this does not replace it.
+ */
+ if (!domain.getServerState().cover(failing))
+ {
+ giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+ }
+ }
+ catch (Throwable cleanupFailure)
+ {
+ logger.error(LocalizableMessage.raw(
+ "the barrier of aChangeParkedByAnUnwoundReplayIsDeliveredAgain could not be given"
+ + " up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
+ }
+ finally
+ {
+ resetReplayGiveUpDelay();
+ }
+ }
+ }
+
+ /**
+ * Gives up on the change no delivery can replay: the budget is shortened to nothing, so
+ * that the change is given up on as soon as one more delivery of it fails. Nothing sends
+ * that change again - it never travelled a session - so its deliveries are made here.
+ */
+ private void giveUpOn(final LDAPReplicationDomain domain, final CSN failing, final DN dn,
+ final List<Modification> mods, final String entryUUID) throws Exception
+ {
+ setReplayGiveUpDelay("0ms");
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(500, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(failing))
+ {
+ domain.processUpdate(new ModifyMsgWhoseOperationRefusesAControl(failing, dn, mods, entryUUID));
+ }
+ assertTrue(domain.getServerState().cover(failing),
+ "the change no delivery can replay must be given up on");
+ }
+ });
+ }
+
+ /**
+ * Test case for [Issue 986]: a change parked as waiting for another one is given back
+ * when the replay thread which parked it is stopped with the pool.
+ * <p>
+ * A parked change stays owned by the replay thread which parked it while that thread
+ * goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
+ * what hands it out again, to whichever replay thread clears the change it was waiting
+ * for. Changing the number of replay threads stops the whole pool and creates another
+ * one, so a thread which parked a change and went back to the queue is joined while it
+ * is idle, and it would end still recorded as the owner of that change - a thread which
+ * does not exist anymore, while every redelivery of a change a replay thread owns is
+ * refused as a duplicate. On a domain which then goes quiet that change is where this
+ * replica's ServerState, and every change behind it from every master, stops.
+ */
+ @Test
+ public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain"));
+
+ final DN waitedOn = addEntryForChange("user.986.1");
+ final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
+ final DN other = addEntryForChange("user.986.2");
+ final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ domain.resetUnreplayedChangeAlertThrottle();
+ final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "no change of this domain is waiting for another one when this test starts");
+
+ final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
+ final CSN failing = gen.newCSN();
+ final CSN parked = gen.newCSN();
+ final CSN applied = gen.newCSN();
+ final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
+ final String parkedDescription = "the change which was parked by a thread the pool stopped";
+ final List<Modification> parkedMods = generatemods("description", parkedDescription);
+ final String appliedDescription = "the change which is applied before the pool is stopped";
+ final List<Modification> appliedMods = generatemods("description", appliedDescription);
+
+ /*
+ * One replay thread, so that the change which is parked is parked by the thread the
+ * pool then stops: a parked change is left owned by the thread which parked it, and
+ * this is about a thread which is not there anymore to be given it back.
+ */
+ setNumUpdateReplayThreads(1);
+ try
+ {
+ /*
+ * A change whose replay failed stays listed and uncommitted - it is what holds this
+ * domain's ServerState back - and stays among the changes the newer ones are checked
+ * against, so a change which follows it on the same entry has to wait for it.
+ */
+ deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
+ () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
+ "the change whose replay fails must stay listed as one which is not in the data");
+
+ /*
+ * The change which is parked as waiting for it. The replay thread it was given to
+ * parks it and goes back to the pool: nothing is waiting for it there, so it is idle
+ * and it still owns the change it parked.
+ */
+ deliverUntilMonitorReaches(domain, "dependent-changes-size", 1,
+ () -> new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID),
+ "a change which waits for one that is not in the data must be parked");
+
+ /*
+ * A change which is applied, on another entry so that it is replayed rather than
+ * parked in its turn. It is there for the count of the deliveries this session took
+ * off it, replayed-updates: a session which is started counts from zero, so that
+ * count going back to zero below is what says the session was restarted - and it is
+ * not left to the give-back alone to put the count above zero before the pool is
+ * stopped. The restart the change whose replay failed asked for is over by now: the
+ * thread which ran it is the one which parked the change above.
+ */
+ domain.processUpdate(new ModifyMsg(applied, other, appliedMods, otherUUID));
+ checkEntryHasAttributeValue(other, "description", appliedDescription, 30,
+ "the change made on the other entry must be applied");
+ // Counted once its ack is out, which is after the change is in the data.
+ final TestTimer counted = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ counted.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(getMonitorAttrValue(baseDN, "replayed-updates") > 0,
+ "the session must have counted the deliveries it took off before the pool is stopped");
+ }
+ });
+
+ /*
+ * How many restarts in a row the session has been through, the barrier's among them:
+ * the restart the give-back of a stopped thread asks for is owed no backoff - what
+ * went away is a replay thread, not the backend - and a restart which waits the
+ * backoff out is the one road which moves this count. It can not move otherwise
+ * between here and the reading below: a replay which made it puts the count back to
+ * zero only once nothing is failing anymore, and the barrier keeps failing until it
+ * is given up, below. The restart the barrier asked for is over: the thread which
+ * ran it is the one which parked the change and applied the other, above.
+ */
+ final int restarts = domain.getConsecutiveSessionRestarts();
+
+ /*
+ * The pool is stopped and created again, the way an administrator changing the
+ * number of replay threads has it: the thread which parked the change is joined
+ * where it waits for the next delivery, and the change it owns is nobody's.
+ */
+ setNumUpdateReplayThreads(2);
+
+ assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 0,
+ "a change parked by a replay thread the pool stopped must be given back");
+ /*
+ * The give-back alone leaves the count where it was, plus the change it handed back:
+ * only a session which is started counts from zero, and the changes of this test
+ * never travelled a session, so nothing is delivered over the one which is started
+ * before the deliveries made below.
+ */
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates", 0,
+ "the session must be restarted for the changes which were given back");
+ assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
+ "the restart run for the changes a stopped thread gave back must not wait the"
+ + " backoff out: what went away is a replay thread, not the backend");
+
+ /*
+ * Nothing sends these changes again - they never travelled a session - so the
+ * deliveries which take over from the ones the stopped pool left behind are made
+ * here. The change no delivery can replay is given up on, which lets the ServerState
+ * past it, and the change which was parked behind it is applied.
+ */
+ setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(500, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ final ServerState state = domain.getServerState();
+ if (!state.cover(failing))
+ {
+ domain.processUpdate(
+ new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID));
+ }
+ if (!state.cover(parked))
+ {
+ domain.processUpdate(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
+ }
+ assertTrue(state.cover(parked),
+ "the change which was given back must be replayed by the delivery which takes it over");
+ }
+ });
+ checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
+ "the change which was parked must be applied by the delivery which took it over");
+ assertEquals(resetNumUpdateReplayThreads(), 0,
+ "the number of replay threads could not be put back");
+ }
+ finally
+ {
+ try
+ {
+ /*
+ * A red before the barrier was given up would leave it listed for the rest of this
+ * class: it never travelled the replication server, so no session restart brings a
+ * delivery which could give it up, and a change which is failing keeps every later
+ * session restart at its backoff and the ServerState behind it. The failure which
+ * is being reported is the one that matters, so this does not replace it.
+ */
+ if (!domain.getServerState().cover(failing))
+ {
+ giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+ }
+ }
+ catch (Throwable cleanupFailure)
+ {
+ logger.error(LocalizableMessage.raw(
+ "the barrier of aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain could not be"
+ + " given up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
+ }
+ finally
+ {
+ try
+ {
+ resetReplayGiveUpDelay();
+ }
+ finally
+ {
+ /*
+ * Best-effort on the way out of a red, with no assertion to replace the failure
+ * which is being reported: on the normal road the number is put back, and checked,
+ * above.
+ */
+ resetNumUpdateReplayThreads();
+ }
+ }
+ }
+ }
+
+ /**
+ * Delivers a change until a monitor attribute of the domain reaches the expected value.
+ * <p>
+ * A delivery is dropped rather than queued while the listener thread is down, which it
+ * is for as long as a recovery is restarting the session, so a change which has to reach
+ * a replay thread is delivered until it does. A delivery of a change a replay thread
+ * owns is refused as the duplicate it is, so the deliveries which follow the one that
+ * was taken cost nothing.
+ */
+ private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain,
+ final String attributeName, final long expected,
+ final Supplier<? extends LDAPUpdateMsg> delivery, final String message) throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(20, SECONDS)
+ .sleepTimes(500, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (getMonitorAttrValue(baseDN, attributeName) != expected)
+ {
+ domain.processUpdate(delivery.get());
+ }
+ assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message);
+ }
+ });
+ }
+
+ /**
+ * Sets how many replay threads this server runs, the way an administrator would: the
+ * pool is stopped and created again with that number.
+ */
+ private static void setNumUpdateReplayThreads(int threads) throws Exception
+ {
+ assertEquals(TestCaseUtils.applyModifications(true,
+ "dn: " + SYNCHRO_PLUGIN_DN,
+ "changetype: modify",
+ "replace: ds-cfg-num-update-replay-threads",
+ "ds-cfg-num-update-replay-threads: " + threads), 0,
+ "the number of replay threads could not be changed");
+ }
+
+ /**
+ * Puts the number of replay threads back to what this server computes for itself, which
+ * is what it runs with when the configuration carries no number of its own.
+ *
+ * @return the result code of the change, so that a finally can call this without an
+ * assertion which would replace the failure it is on the way out of
+ */
+ private static int resetNumUpdateReplayThreads() throws Exception
+ {
+ return TestCaseUtils.applyModifications(true,
+ "dn: " + SYNCHRO_PLUGIN_DN,
+ "changetype: modify",
+ "delete: ds-cfg-num-update-replay-threads");
+ }
+
+ /** Adds the entry a change of these tests is made on. */
+ private DN addEntryForChange(String uid) throws Exception
+ {
+ return TestCaseUtils.addEntry(
+ "dn: uid=" + uid + "," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: " + uid,
+ "cn: Aaccf Amar",
+ "sn: Amar").getName();
+ }
+
/** A delivery of a change whose replay does not run to its end. */
private interface FailingDelivery
{
@@ -3603,16 +4106,7 @@
private void assertChangeIsDeliveredAgainAfter(
FailingDelivery delivery, int serverId, String uid, final String description) throws Exception
{
- Entry tmp = TestCaseUtils.addEntry(
- "dn: uid=" + uid + "," + baseDN,
- "objectClass: top",
- "objectClass: person",
- "objectClass: organizationalPerson",
- "objectClass: inetOrgPerson",
- "uid: " + uid,
- "cn: Aaccf Amar",
- "sn: Amar");
- final DN dn = tmp.getName();
+ final DN dn = addEntryForChange(uid);
final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
@@ -3842,6 +4336,36 @@
}
/**
+ * A ModifyMsg whose ack runs out of memory on the way out of a replay which applied it.
+ * <p>
+ * The change is committed before the ack of its delivery is published, and the ack is
+ * where the JVM runs out of memory: the one throw from there which is not caught, so the
+ * replay is unwound with the change it was replaying in the data and owned by nobody -
+ * commit() cleared the owner - and the thread ends on the error. What the give-back on
+ * the way out of {@code replay()} has left to give back is the changes this thread parked
+ * as waiting for another one, and the restart it asks for them is the one which is run:
+ * the road a failed replay takes to ask for its own change again is not on the way.
+ * <p>
+ * Nothing on the way in reads what throws here: a message handed to the domain rather
+ * than published is not one this server acknowledges to anybody.
+ */
+ private static final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
+ {
+ private ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ // Read first thing by processUpdateDone(), which is what publishes the ack.
+ throw new OutOfMemoryError("the ack of this applied delivery runs out of memory");
+ }
+ }
+
+ /**
* An AddMsg whose ack throws on the way out of a replay which applied it.
* <p>
* The change is committed before the ack of its delivery is published, so this is the
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ForEachDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ForEachDomainTest.java
new file mode 100644
index 0000000..d3b3c91
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ForEachDomainTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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 org.assertj.core.api.Assertions.*;
+import static org.opends.server.TestCaseUtils.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.backends.MemoryBackend;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Tests that {@link MultimasterReplication#forEachDomain} gives every domain its turn
+ * whatever one of them threw, and reports the failures the way it says it does: the first
+ * is thrown once the loop is over, the others are suppressed under it where it records
+ * suppression, and an instance thrown by two domains - the error a JVM out of memory hands
+ * out as often as it is asked for one - is thrown once and suppresses nothing (issue #986).
+ * <p>
+ * Two domains, on backends of their own, neither of them started: the action is what
+ * throws, and it never touches the domain it is given.
+ */
+@SuppressWarnings("javadoc")
+public class ForEachDomainTest extends ReplicationTestCase
+{
+ private static final int RS_ID = 613;
+ private static final String SECOND_BACKEND_ID = "test2";
+ private static final String SECOND_ROOT_DN_STRING = "o=" + SECOND_BACKEND_ID;
+
+ private DN firstBaseDN;
+ private DN secondBaseDN;
+ private ReplicationServer replicationServer;
+ private LDAPReplicationDomain firstDomain;
+ private LDAPReplicationDomain secondDomain;
+
+ @BeforeMethod
+ public void setUpLocal() throws Exception
+ {
+ firstBaseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ secondBaseDN = DN.valueOf(SECOND_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ TestCaseUtils.initializeMemoryBackend(SECOND_BACKEND_ID, SECOND_ROOT_DN_STRING, true);
+
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ rsPort, "forEachDomainTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
+ final SortedSet<String> replServers = new TreeSet<>();
+ replServers.add("localhost:" + rsPort);
+ firstDomain = MultimasterReplication.createNewDomain(new DomainFakeCfg(firstBaseDN, 1, replServers));
+ secondDomain = MultimasterReplication.createNewDomain(new DomainFakeCfg(secondBaseDN, 2, replServers));
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception
+ {
+ try
+ {
+ MultimasterReplication.deleteDomain(firstBaseDN);
+ MultimasterReplication.deleteDomain(secondBaseDN);
+ }
+ finally
+ {
+ try
+ {
+ remove(replicationServer);
+ }
+ finally
+ {
+ final MemoryBackend backend = (MemoryBackend) getServerContext().getBackendConfigManager()
+ .getLocalBackendById(SECOND_BACKEND_ID);
+ if (backend != null)
+ {
+ backend.clearMemoryBackend();
+ backend.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(backend);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void theFirstFailureIsThrownOnceTheLoopIsOverWithTheOthersSuppressedUnderIt()
+ {
+ final RuntimeException first = new RuntimeException("first");
+ final RuntimeException second = new RuntimeException("second");
+ final List<LDAPReplicationDomain> visited = new ArrayList<>();
+
+ final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
+ {
+ visited.add(domain);
+ throw visited.size() == 1 ? first : second;
+ }));
+
+ assertThat(thrown).as("the failure thrown once the loop is over must be the first one met")
+ .isSameAs(first);
+ assertThat(thrown.getSuppressed()).as("the failure met after the first must be suppressed under it")
+ .containsExactly(second);
+ assertThat(visited).as("every domain must get its turn whatever the one before it threw")
+ .containsExactlyInAnyOrder(firstDomain, secondDomain);
+ }
+
+ @Test
+ public void aFailureThrownByTwoDomainsIsThrownOnceAndSuppressesNothing()
+ {
+ // The error a JVM out of memory prepared beforehand is handed out to every domain alike.
+ final OutOfMemoryError theOneError = new OutOfMemoryError("prepared beforehand");
+ final List<LDAPReplicationDomain> visited = new ArrayList<>();
+
+ final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
+ {
+ visited.add(domain);
+ throw theOneError;
+ }));
+
+ assertThat(thrown).as("the failure thrown once the loop is over must be the one both domains threw")
+ .isSameAs(theOneError);
+ assertThat(thrown.getSuppressed()).as("a failure must not be suppressed under itself").isEmpty();
+ assertThat(visited).as("every domain must get its turn whatever the one before it threw")
+ .containsExactlyInAnyOrder(firstDomain, secondDomain);
+ }
+
+ @Test
+ public void aFirstFailureWhichRecordsNoSuppressionStillGivesEveryDomainItsTurn()
+ {
+ final Error first = new ErrorWhichRecordsNoSuppression();
+ final RuntimeException second = new RuntimeException("second");
+ final List<LDAPReplicationDomain> visited = new ArrayList<>();
+
+ final Throwable thrown = catchThrowable(() -> MultimasterReplication.forEachDomain(domain ->
+ {
+ visited.add(domain);
+ if (visited.size() == 1)
+ {
+ throw first;
+ }
+ throw second;
+ }));
+
+ assertThat(thrown).as("the failure thrown once the loop is over must be the first one met")
+ .isSameAs(first);
+ assertThat(thrown.getSuppressed()).as("a failure which records no suppression must have none recorded")
+ .isEmpty();
+ assertThat(visited).as("every domain must get its turn whatever the one before it threw")
+ .containsExactlyInAnyOrder(firstDomain, secondDomain);
+ }
+
+ /**
+ * An error which keeps no record of the failures suppressed under it: the shape of the
+ * error a JVM out of memory prepared beforehand, which was made without its constructor
+ * and so keeps no list to record them in.
+ */
+ private static final class ErrorWhichRecordsNoSuppression extends Error
+ {
+ private static final long serialVersionUID = 1L;
+
+ private ErrorWhichRecordsNoSuppression()
+ {
+ super("records no suppression", null, false, false);
+ }
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied.java
new file mode 100644
index 0000000..2c9cf8a
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied.java
@@ -0,0 +1,82 @@
+/*
+ * 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.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.types.Modification;
+
+/**
+ * A ModifyMsg whose ack runs out of memory on the way out of a replay which applied it.
+ * <p>
+ * The change is committed before the ack of its delivery is published, and the ack is
+ * where the JVM runs out of memory: the one throw from there which is not caught, so the
+ * replay is unwound with the change it was replaying in the data and owned by nobody -
+ * {@code commit()} cleared the owner - and the thread ends on the error. What the give-back
+ * on the way out of {@code replay()} has left to give back is the changes this thread
+ * parked as waiting for another one, and the restart it asks for them is the one which is
+ * run: the road a failed replay takes to ask for its own change again is not on the way.
+ * <p>
+ * Given a flag, the ack sets it before it throws. Handed the flag the replay reads as "this
+ * thread is stopping", that stops the thread while its change is being acknowledged: applied
+ * and committed under a running thread, unwound under a stopping one, the way a thread of
+ * the pool is stopped when their number is changed while its ack is on its way out. A flag
+ * set before the replay would not reach here with the change applied: the replay abandons
+ * the change unapplied at the top of its first attempt, still owned by this thread, and the
+ * abandon road asks for a restart of its own next to the one the give-back asks for.
+ * <p>
+ * Nothing on the way in reads what throws here: a message handed to the domain rather than
+ * published is not one this server acknowledges to anybody. The twin of the fixture
+ * {@code UpdateOperationTest} unwinds a replay thread with.
+ */
+final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
+{
+ /** The flag the ack sets before it throws, or {@code null} for none. */
+ private final AtomicBoolean setByTheAck;
+
+ ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ this(csn, dn, mods, entryUUID, null);
+ }
+
+ /**
+ * @param setByTheAck the flag the ack sets before it throws: the one the replay reads as
+ * "this thread is stopping", to stop the thread while its applied change is
+ * being acknowledged
+ */
+ ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID, AtomicBoolean setByTheAck)
+ {
+ super(csn, dn, mods, entryUUID);
+ this.setByTheAck = setByTheAck;
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ // Read first thing by processUpdateDone(), which is what publishes the ack.
+ if (setByTheAck != null)
+ {
+ setByTheAck.set(true);
+ }
+ throw new OutOfMemoryError("the ack of this applied delivery runs out of memory");
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationRefusesAControl.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationRefusesAControl.java
new file mode 100644
index 0000000..9db7874
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationRefusesAControl.java
@@ -0,0 +1,67 @@
+/*
+ * 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 org.opends.server.protocols.internal.InternalClientConnection.*;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.forgerock.opendj.ldap.ByteString;
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.core.ModifyOperation;
+import org.opends.server.core.ModifyOperationBasis;
+import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.protocol.ModifyContext;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.replication.protocol.OperationContext;
+import org.opends.server.types.Control;
+import org.opends.server.types.Modification;
+import org.opends.server.types.RawModification;
+
+/**
+ * A ModifyMsg whose operation can not be prepared for its replay.
+ * <p>
+ * The operation is built - so the replay is past the point where a message is given up on -
+ * and the list of request controls it carries can not be added to, so the ManageDsaIT
+ * control the replay puts on every operation throws before
+ * {@code OperationContext.getCSN(op)} is reached: the change fails, stays listed and
+ * uncommitted, and is asked for again. Such a message can not travel the protocol:
+ * {@code ModifyMsg.createOperation()} builds an operation whose controls can be added to,
+ * so this one is handed to the domain rather than published. The twin of the fixture
+ * {@code UpdateOperationTest} holds its barrier with.
+ */
+final class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
+{
+ ModifyMsgWhoseOperationRefusesAControl(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
+ {
+ final ModifyOperation op = new ModifyOperationBasis(connection, nextOperationID(),
+ nextMessageID(), Collections.<Control>emptyList(),
+ ByteString.valueOfUtf8(getDN().toString()), new ArrayList<RawModification>());
+ op.setAttachment(OperationContext.SYNCHROCONTEXT,
+ new ModifyContext(getCSN(), getEntryUUID()));
+ return op;
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ParkedChangeGiveBackTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ParkedChangeGiveBackTest.java
new file mode 100644
index 0000000..614b021
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ParkedChangeGiveBackTest.java
@@ -0,0 +1,373 @@
+/*
+ * 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 org.assertj.core.api.Assertions.*;
+import static org.opends.messages.ReplicationMessages.*;
+import static org.opends.server.TestCaseUtils.*;
+import static org.testng.Assert.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
+import org.opends.server.TestCaseUtils;
+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.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.replication.protocol.UpdateMsg;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.types.Entry;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Tests who runs the session restart asked for the changes a replay which is unwound had
+ * parked as waiting for another change (issue #954).
+ * <p>
+ * The thread which gave them back runs it itself, so that the delivery they wait for is
+ * not left to the next tick of the state checkpointer - unless that thread is stopping, in
+ * which case it leaves the request standing the way a stopping thread leaves the request it
+ * makes for the change it abandons: the threads of the pool are stopped one after the
+ * other and joined, and each running a restart on its way out would have the configuration
+ * change which is stopping them wait for one restart per thread.
+ * <p>
+ * The replay runs on the thread of the test, which is what says whether that thread is
+ * stopping, and the error which unwinds it is caught here rather than ending a replay
+ * thread. The thread which is stopping is stopped by the ack of the change it applied, and
+ * the case asserts that the change was applied: stopped before the replay, it abandons the
+ * change unapplied at the top of its first attempt, and the abandon road asks for a restart
+ * of its own next to the give-back's, which would then be pinned by nothing - that road is
+ * the third case's, which pins the abandon arm of the catch. The restart is asked to fail
+ * once, so that the thread which ran it is the one which met the failure: a replay thread
+ * carries the failure out with the error it is ending on, the state checkpointer reports
+ * it. The restart is asked for at once, without the backoff, and the count of the restarts
+ * in a row says so: only the restart run again after the failure waits its backoff out.
+ */
+@SuppressWarnings("javadoc")
+public class ParkedChangeGiveBackTest extends ReplicationTestCase
+{
+ private static final int RS_ID = 612;
+ private static final int DS_ID = 1;
+ private static final AtomicBoolean RUNNING = new AtomicBoolean(false);
+
+ /**
+ * How long the state checkpointer is given to run a request left standing and report
+ * the failure it was asked to meet: its next tick, at most a second away, and the
+ * report.
+ */
+ private static final long CHECKPOINTER_BOUND_IN_MS = 5000;
+ /**
+ * How long the session is given to come back once a restart failed: the backoff a
+ * restart which follows a failed one is owed - the second in a row here, two seconds -
+ * and the tick of the state checkpointer which runs it.
+ */
+ private static final long RESTART_BOUND_IN_MS = 10000;
+
+ private DN baseDN;
+ private ReplicationServer replicationServer;
+ private LDAPReplicationDomain domain;
+ private TestSynchronousReplayQueue queue;
+ private CSNGenerator gen;
+
+ @BeforeMethod
+ public void setUpLocal() throws Exception
+ {
+ baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ rsPort, "parkedChangeGiveBackTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
+
+ final SortedSet<String> replServers = new TreeSet<>();
+ replServers.add("localhost:" + rsPort);
+ final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID, replServers);
+ conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);
+ queue = new TestSynchronousReplayQueue();
+ domain = MultimasterReplication.createNewDomain(conf, queue);
+ domain.start();
+ assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
+ gen = new CSNGenerator(201, 0);
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception
+ {
+ try
+ {
+ domain.failNextSessionRestarts(0);
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+ finally
+ {
+ remove(replicationServer);
+ }
+ }
+
+ /**
+ * The thread which gave a parked change back runs the restart it asked for, and meets
+ * the failure that restart was asked to meet: the failure comes out with the error which
+ * unwound the replay, as one it suppressed, and the restart which is run again after it
+ * brings the session back before the replay returns. The restart it asked for is run at
+ * once: the one run again after the failure is the one which waits its backoff out, and
+ * that is the one move of the count of the restarts in a row.
+ */
+ @Test(timeOut = 120_000)
+ public void theThreadWhichGaveBackAParkedChangeRunsTheRestartItAskedFor() throws Exception
+ {
+ parkAChangeBehindABarrier(addEntry("waitedOn"));
+ final Entry other = addEntry("other");
+ final int restartsBefore = domain.getConsecutiveSessionRestarts();
+ domain.failNextSessionRestarts(1);
+
+ final OutOfMemoryError unwinding = unwindAReplay(other, Stopped.NEVER);
+
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "the change parked by the replay which was unwound must be given back");
+ assertThat(injectedRestartFailuresAmong(unwinding.getSuppressed()))
+ .as("the thread which gave the parked change back must have run the restart it"
+ + " asked for, and met the failure that restart was asked to meet")
+ .hasSize(1);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 0,
+ "the restart which was asked to fail never ran");
+ awaitConnected(RESTART_BOUND_IN_MS, "the restart run again after the one which failed did"
+ + " not bring the session back");
+ assertEquals(domain.getConsecutiveSessionRestarts(), restartsBefore + 1,
+ "the restart the give-back asked for must be run at once, without the backoff: the one"
+ + " run again after the failure is the one which waits it out");
+ }
+
+ /**
+ * A thread which is stopping leaves the restart it asked for standing, and the state
+ * checkpointer runs it: the failure that restart was asked to meet is the checkpointer's
+ * to report, and nothing of it comes out with the error which unwound the replay.
+ * <p>
+ * The thread is stopped by the ack of the change it applied, so the change is committed
+ * and owned by nobody by the time the replay is unwound, and the request the give-back
+ * makes for the parked change is the one request which stands for the checkpointer. The
+ * case asserts that road: a thread stopped before the replay hands its change back on a
+ * road of its own, which asks for a restart next to the give-back's and would stand in
+ * for it.
+ */
+ @Test(timeOut = 120_000)
+ public void aStoppingThreadLeavesTheRestartItAskedForToTheStateCheckpointer() throws Exception
+ {
+ parkAChangeBehindABarrier(addEntry("waitedOn"));
+ final Entry other = addEntry("other");
+ final int reportedBefore = restartFailureReports().size();
+ final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
+ final int restartsBefore = domain.getConsecutiveSessionRestarts();
+ domain.failNextSessionRestarts(1);
+
+ final OutOfMemoryError unwinding = unwindAReplay(other, Stopped.BY_THE_ACK);
+
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore + 1,
+ "the change must be applied and its replay unwound by the ack, not abandoned unapplied");
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "the change parked by the replay which was unwound must be given back");
+ assertThat(injectedRestartFailuresAmong(unwinding.getSuppressed()))
+ .as("a thread which is stopping must not run the restart it asked for")
+ .isEmpty();
+
+ final long deadline = System.currentTimeMillis() + CHECKPOINTER_BOUND_IN_MS;
+ while (restartFailureReports().size() == reportedBefore)
+ {
+ assertTrue(System.currentTimeMillis() < deadline, "the state checkpointer did not run"
+ + " the restart the stopping thread left standing: the failure that restart was"
+ + " asked to meet was never reported");
+ Thread.sleep(50);
+ }
+ awaitConnected(RESTART_BOUND_IN_MS,
+ "the session was not brought back after the restart which failed");
+ assertEquals(domain.getConsecutiveSessionRestarts(), restartsBefore + 1,
+ "the restart the give-back asked for must be run at once, without the backoff: the one"
+ + " run again after the failure is the one which waits it out");
+ }
+
+ /**
+ * A thread stopped before its first attempt hands back the change it did not apply, the
+ * way a stopping thread abandons a replay: the ack of an abandoned change is published
+ * all the same, and it is what runs out of memory here, so the abandon road taken is the
+ * one of the catch - the change is still this thread's when the replay is unwound.
+ */
+ @Test(timeOut = 120_000)
+ public void aStoppingThreadHandsBackTheChangeItDidNotApply() throws Exception
+ {
+ parkAChangeBehindABarrier(addEntry("waitedOn"));
+ final Entry other = addEntry("other");
+ final long appliedBefore = getMonitorAttrValue(baseDN, "replayed-updates-ok");
+ final CSN abandoned = gen.newCSN();
+
+ unwindAReplay(other, abandoned, Stopped.BEFORE_THE_REPLAY);
+
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-ok"), appliedBefore,
+ "a thread stopped before the replay must not apply the change");
+ assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), abandoned))
+ .as("a stopping thread must hand back the change it did not apply")
+ .isNotEmpty();
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "the change parked by the thread which is stopping must be given back");
+ }
+
+ private void awaitConnected(long boundInMs, String message) throws Exception
+ {
+ final long deadline = System.currentTimeMillis() + boundInMs;
+ while (!domain.isConnected())
+ {
+ assertTrue(System.currentTimeMillis() < deadline, message);
+ Thread.sleep(50);
+ }
+ }
+
+ private Entry addEntry(String cn) throws Exception
+ {
+ return TestCaseUtils.addEntry(
+ "dn: cn=" + cn + "," + TEST_ROOT_DN_STRING,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: " + cn,
+ "sn: " + cn);
+ }
+
+ /**
+ * Replays, on the thread of this test, a change whose replay fails and stays listed - its
+ * operation is built and then refused, so it is asked for again rather than stepped over -
+ * and then a change on the same entry, which is parked as waiting for it and owned by this
+ * thread from then on. The restart the failed change asks for is run on this thread as
+ * well, so the session is back once this returns.
+ */
+ private void parkAChangeBehindABarrier(Entry entry) throws Exception
+ {
+ final String entryUUID = getEntryUUID(entry.getName());
+ final CSN failing = gen.newCSN();
+ replayMsg(new ModifyMsgWhoseOperationRefusesAControl(failing, entry.getName(),
+ generatemods("description", "the replay of this change fails"), entryUUID), RUNNING);
+ assertFalse(domain.getServerState().cover(failing),
+ "the change whose replay fails must stay listed as one which is not in the data");
+ awaitConnected(RESTART_BOUND_IN_MS,
+ "the session was not brought back for the change whose replay failed");
+
+ replayMsg(new ModifyMsg(gen.newCSN(), entry.getName(),
+ generatemods("description", "the change which was parked as a dependency"), entryUUID),
+ RUNNING);
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
+ "a change which waits for one that is not in the data must be parked");
+ }
+
+ /** When the thread a replay runs on is stopped, if it is. */
+ private enum Stopped
+ {
+ /** It is running throughout. */
+ NEVER,
+ /** By the ack of the change it applied: the change is committed, and the replay is unwound. */
+ BY_THE_ACK,
+ /**
+ * Before the replay: the change is abandoned unapplied at the top of its first attempt,
+ * and the ack which says so is what runs out of memory.
+ */
+ BEFORE_THE_REPLAY
+ }
+
+ private OutOfMemoryError unwindAReplay(Entry entry, Stopped stopped) throws Exception
+ {
+ return unwindAReplay(entry, gen.newCSN(), stopped);
+ }
+
+ /**
+ * Replays, on the thread of this test, a change whose ack runs out of memory, and returns
+ * the error the replay was unwound on.
+ *
+ * @param stopped when the thread the replay runs on is stopped: running until the ack,
+ * the change is applied rather than abandoned at the top of its first attempt
+ */
+ private OutOfMemoryError unwindAReplay(Entry entry, CSN csn, Stopped stopped) throws Exception
+ {
+ // Fresh for every replay: the ack is what sets it, and nothing puts it back.
+ final AtomicBoolean stopping = new AtomicBoolean(stopped == Stopped.BEFORE_THE_REPLAY);
+ try
+ {
+ replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(csn, entry.getName(),
+ generatemods("description", "the replay of this change is unwound by its ack"),
+ getEntryUUID(entry.getName()), stopped == Stopped.BY_THE_ACK ? stopping : null), stopping);
+ }
+ catch (OutOfMemoryError unwinding)
+ {
+ // The error is the fixture's own, and this is the thread it would have ended.
+ return unwinding;
+ }
+ throw new AssertionError("the replay was not unwound: the ack of the delivery must run out of memory");
+ }
+
+ /** The throwables among the provided ones which a restart threw because a test asked it to. */
+ private static List<Throwable> injectedRestartFailuresAmong(Throwable[] suppressed)
+ {
+ final List<Throwable> injected = new ArrayList<>();
+ for (Throwable t : suppressed)
+ {
+ if (t instanceof IllegalStateException && t.getMessage().contains("as a test asked"))
+ {
+ injected.add(t);
+ }
+ }
+ return injected;
+ }
+
+ /** The records of the error log which report a session restart of this domain that threw. */
+ private List<String> restartFailureReports()
+ {
+ final List<String> records = new ArrayList<>();
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ if (record.contains("msgID=" + ERR_REPLAY_SESSION_RESTART_FAILED.ordinal())
+ && record.contains(baseDN.toString()))
+ {
+ records.add(record);
+ }
+ }
+ return records;
+ }
+
+ /** The records of the error log which carry the provided message for the provided change. */
+ private static List<String> errorLogRecordsOf(int msgId, CSN csn)
+ {
+ final List<String> records = new ArrayList<>();
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ if (record.contains("msgID=" + msgId) && record.contains(csn.toString()))
+ {
+ records.add(record);
+ }
+ }
+ return records;
+ }
+
+ private void replayMsg(UpdateMsg updateMsg, AtomicBoolean stopping) throws InterruptedException
+ {
+ domain.processUpdate(updateMsg);
+ final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage();
+ domain.markInProgress(ldapUpdate);
+ domain.replay(ldapUpdate, stopping);
+ }
+}
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 e91292f..30107ac 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
@@ -15,6 +15,7 @@
*/
package org.opends.server.replication.plugin;
+import static java.util.Collections.*;
import static org.testng.Assert.*;
import java.util.NoSuchElementException;
@@ -670,10 +671,12 @@
/**
* The change a thread parked as waiting for another one is not the change it is
- * replaying: it is handed to whichever thread clears what it waits for, so a give-back on
- * the way out of an unwound replay must leave it alone. Releasing it without taking it out
- * of the changes which are waiting would have the same change handed to two threads
- * (issue #922).
+ * replaying: it is handed to whichever thread clears what it waits for, so the give-back
+ * of the change a replay was unwound on must leave it alone. Releasing it without taking
+ * it out of the changes which are waiting would have the same change handed to two
+ * threads (issue #922) - which is why the parked ones are given back on a road of their
+ * own, {@link RemotePendingChanges#releaseParkedChangesOwnedByCurrentThread()}, where
+ * both happen in one step (issue #954).
* <p>
* The deliveries are taken in the order a replay thread takes them: one at a time, off
* the queue the pool shares. So the change which is parked here is parked by the thread
@@ -799,6 +802,145 @@
"a change which has been handed out must not be handed out again");
}
+ /**
+ * Test case for [Issue 954]: a change parked as waiting for another one is given back
+ * when the replay which parked it is unwound.
+ * <p>
+ * A parked change stays owned by the thread which parked it, and is handed out again by
+ * {@link RemotePendingChanges#getNextUpdate()} to whichever thread clears the change it
+ * waits for. A replay which is unwound - a JVM out of memory, a throw from what the replay
+ * runs once the ack of its delivery is out - leaves that thread without a road back to the
+ * change: it takes the next delivery off the queue instead. Nothing else asks for the
+ * change either, since every redelivery of a change a replay thread owns is refused as a
+ * duplicate, so this domain's ServerState would stay behind it until some other change is
+ * replayed on this domain.
+ */
+ @Test
+ public void aChangeParkedByAReplayWhichIsUnwoundIsGivenBack() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN deleted = generator.newCSN();
+ final CSN renamed = generator.newCSN();
+
+ final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(delete));
+ assertTrue(pendingChanges.markInProgress(delete));
+
+ // A rename into the DN that delete is on: it can only be replayed once the delete has been.
+ final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
+ assertTrue(pendingChanges.putRemoteUpdate(rename));
+ assertTrue(pendingChanges.markInProgress(rename));
+ assertTrue(pendingChanges.checkDependencies(rename));
+
+ // The replay which parked it is unwound, so it gives back what it still owns.
+ assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(),
+ singletonList(renamed), "the change this thread parked must be given back");
+
+ assertEquals(pendingChanges.getDependentChangesSize(), 0,
+ "a change which was given back must not be left waiting for a thread to hand it out");
+ assertEquals(pendingChanges.getQueueSize(), 2, "the change must stay listed as pending");
+ assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed");
+ assertEquals(pendingChanges.changesInProgressSize(), 2,
+ "a change which is not in the data yet must stay a dependency of the changes which follow it");
+
+ assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
+ "the change the unwound replay gave back must be taken over by the next delivery");
+ }
+
+ /**
+ * Test case for [Issue 954]: a change which was given back is handed out by the next
+ * delivery of it and by nothing else.
+ * <p>
+ * Unparking it and giving it back is one step under both locks for that reason: a
+ * change which was released while it is still listed as waiting would be replayed by
+ * the thread {@link RemotePendingChanges#getNextUpdate()} hands it to and by the thread
+ * which takes over the delivery which follows - the double replay the ownership is
+ * there to prevent (OPENDJ-1115).
+ */
+ @Test
+ public void aParkedChangeWhichWasGivenBackIsNotHandedOutAsADependency() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN deleted = generator.newCSN();
+ final CSN renamed = generator.newCSN();
+
+ final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(delete));
+ assertTrue(pendingChanges.markInProgress(delete));
+
+ final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
+ assertTrue(pendingChanges.putRemoteUpdate(rename));
+ assertTrue(pendingChanges.markInProgress(rename));
+ assertTrue(pendingChanges.checkDependencies(rename));
+ pendingChanges.releaseParkedChangesOwnedByCurrentThread();
+
+ // The change it was waiting for is replayed, which is what used to hand it out.
+ pendingChanges.commit(deleted);
+
+ assertNull(pendingChanges.getNextUpdate(),
+ "a change which was given back must not also be handed out as a dependency");
+
+ // It is replayed by the delivery which takes it over, and by that one only.
+ final ModifyDNMsg nextDelivery = renameIntoDeletedEntry(renamed);
+ assertTrue(pendingChanges.putRemoteUpdate(nextDelivery));
+ assertTrue(pendingChanges.markInProgress(nextDelivery));
+ pendingChanges.commit(renamed);
+
+ assertTrue(state.cover(renamed));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * Test case for [Issue 954]: the changes another thread parked are left alone.
+ * <p>
+ * A change is given back by the thread which owns it and by nobody else, here as
+ * everywhere else (issue #922). The thread which parked a change is the one which may
+ * still be inside the dependency checks which parked it - they park a change once per
+ * dependency it has - so a change released under it would be listed as waiting again a
+ * moment later, and handed out while the delivery which took it over is being replayed.
+ */
+ @Test
+ public void theParkedChangesOfAnotherThreadAreLeftAlone() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN deleted = generator.newCSN();
+ final CSN renamed = generator.newCSN();
+
+ final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(delete));
+ assertTrue(pendingChanges.markInProgress(delete));
+
+ final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
+ assertTrue(pendingChanges.putRemoteUpdate(rename));
+ assertTrue(pendingChanges.markInProgress(rename));
+ assertTrue(pendingChanges.checkDependencies(rename));
+
+ // Another replay thread is unwound while this one holds the change it parked.
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), emptyList(),
+ "a change another thread parked is not this one's to give back");
+ }
+ });
+
+ assertEquals(pendingChanges.getDependentChangesSize(), 1,
+ "the change must stay listed as waiting for the one it depends on");
+ assertFalse(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
+ "a change a replay thread owns must not be taken over (OPENDJ-1115)");
+
+ // It is still handed to whichever thread clears the change it was waiting for.
+ pendingChanges.commit(deleted);
+ assertSame(pendingChanges.getNextUpdate(), rename);
+ }
+
/** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */
private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception
{
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 0f0c15c..23d4f01 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
@@ -50,6 +50,7 @@
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
+import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -66,7 +67,9 @@
* 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 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).
* <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
@@ -293,6 +296,115 @@
}
/**
+ * 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.
+ * <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,
+ * and the backend is live meanwhile, so the change which is parked and the replay which
+ * is unwound run as they would on any domain. The domain going away, or being disabled,
+ * forgets its pending changes a moment after it takes the session and clears every
+ * request and every count on its way, so a give-back on that road is a race with the
+ * forgetting and leaves nothing to read.
+ * <p>
+ * The replay is unwound on the thread of this test - it applied its change, and the ack
+ * of its delivery runs out of memory - so the parked change is this thread's to give
+ * back, and the error which ends a replay thread is caught here instead.
+ */
+ @Test(timeOut = 120_000)
+ public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain() 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 String[] exported = exportedEntries();
+
+ // The request is out, and the exporter holds it until the give-back below has run.
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ /*
+ * The barrier: a change whose replay fails stays listed and uncommitted - the attempts
+ * in place end on an entryUUID search which does not run, the way they do in the case
+ * above - and stays among the changes the newer ones are checked against, so a change
+ * which follows it on the same entry has to wait for it. The restart which would have
+ * followed is refused, the total update owning the session, and the search is let
+ * through again before anything below reads a monitor.
+ */
+ final DN movedAway = DN.valueOf("cn=movedAway," + EXAMPLE_DN);
+ final CSN failing = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new ModifyMsg(failing, movedAway,
+ generatemods("description", "the replay of this change fails"), entryUUID));
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+ assertFalse(domain.getServerState().cover(failing),
+ "the change whose replay fails must stay listed as one which is not in the data");
+
+ // Parked as waiting for it by this thread, which owns it from here on.
+ final CSN parked = gen.newCSN();
+ replayMsg(new ModifyMsg(parked, movedAway,
+ generatemods("description", "the change which was parked as a dependency"), entryUUID));
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
+ "a change which waits for one that is not in the data must be parked");
+
+ /*
+ * The replay which is unwound while this thread still holds the parked change: its own
+ * change is applied and committed, so the give-back on the way out finds the parked
+ * change alone. The count is read once the parked change is listed, since a parked
+ * change publishes no ack and is not counted until the delivery which replays it is.
+ */
+ final long processed = getMonitorAttrValue(baseDN, "replayed-updates");
+ final CSN unwound = gen.newCSN();
+ try
+ {
+ replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, entry.getName(),
+ generatemods("description", "the replay of this change is unwound once it is applied"),
+ entryUUID));
+ 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(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");
+ 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")
+ .isEmpty();
+ assertTrue(domain.isConnected(), "the session the answer to the request arrives over was stopped");
+
+ answerImportRequest(exported.length);
+ finishImport(exported);
+ for (String ldif : exported)
+ {
+ final DN dn = dnOf(ldif);
+ assertTrue(entryExists(dn), "the import ended before " + dn
+ + " arrived: the answer to the request was lost with the session it was made over");
+ }
+ }
+
+ /**
* 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:
--
Gitblit v1.10.0