From 776339a8c63c0bd8c83cfe8a619f6146794358e8 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Sat, 12 Sep 2026 11:53:58 +0000
Subject: [PATCH] [#922] Give a change back when the replay which owns it is unwound (#958)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 186 ++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 11
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java | 37
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java | 24
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 977 ++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 382 ++++++++++
opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java | 131 +++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java | 372 ++++++++++
8 files changed, 2,067 insertions(+), 53 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 64eded8..b94a4dc 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
@@ -338,6 +338,17 @@
* in flight unreplayable, and one alert per change would be a storm.
*/
private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000;
+ /**
+ * What the ack of a delivery whose replay ran out of memory says did not apply the
+ * change.
+ * <p>
+ * A constant rather than a message built from the change, because this is read on the
+ * road out of an {@code OutOfMemoryError}: formatting one asks the JVM for the memory
+ * it has just refused, and what {@code processUpdateDone()} needs of this string is
+ * that there is one - it sets {@code hasReplayError} on the ack and never sends the
+ * text, which is why no ordinal is spent on it either.
+ */
+ private static final String REPLAY_RAN_OUT_OF_MEMORY = "the replay of this change ran out of memory";
/** The number of updates this replica gave up replaying. */
private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger();
/** Set while a replay thread is restarting the session after a failed replay. */
@@ -2564,6 +2575,11 @@
/**
* Create and replay a synchronized Operation from an UpdateMsg.
+ * <p>
+ * The change is given back on the way out of a replay which was unwound before it
+ * reached one of the roads which give it back: a change which stays owned by a thread
+ * which is not replaying it anymore is refused as a duplicate on every later delivery,
+ * so this domain's ServerState would never move past it (issue #922).
*
* @param msg
* The UpdateMsg to be replayed.
@@ -2572,6 +2588,153 @@
*/
void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
{
+ try
+ {
+ replayChangeAndTheChangesWaitingForIt(msg, replayThreadShutdown);
+ }
+ catch (Throwable t)
+ {
+ /*
+ * The roads which run to their end give the change back themselves, so what is left
+ * to give back here is a change whose replay was unwound over them: by an
+ * OutOfMemoryError, which is left to end this thread, or by a throw from what the
+ * replay runs once the ack of the delivery has been published - the give-back of the
+ * change and the hand-out of the changes which were waiting for it are on that road.
+ *
+ * It takes the road of a failed replay rather than being handed back on the spot, so
+ * that a change which keeps unwinding the replays it is given to is counted as
+ * 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.
+ *
+ * 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.
+ */
+ CSN owned = null;
+ try
+ {
+ owned = remotePendingChanges.getChangeOwnedByCurrentThread();
+ if (owned != null)
+ {
+ if (replayThreadShutdown.get() || shutdown.get() || disabled)
+ {
+ /*
+ * The replay was not failing, it was being abandoned: this thread is stopping
+ * because their number is being changed, or the domain is going away or is
+ * being imported into. The change is handed back without being counted against
+ * its give-up budget, which is the road a replay abandoned that way takes.
+ */
+ abandonReplay(owned);
+ }
+ else
+ {
+ /*
+ * A JVM which has run out of memory is told apart here rather than reported on
+ * its own road: the change is given back counted, the way every other unwound
+ * replay gives it back, but the line which says it is being asked for again is
+ * not built - that asks the JVM for the memory it has just refused - and the
+ * session is restarted without sitting through the backoff, since the thread
+ * which is doing it is on its way out.
+ *
+ * A change whose budget is spent is still reported and still raises its alert
+ * on this road: it is the one line which says this replica has diverged, and an
+ * operator who is not told would be left with a replica which is silently
+ * behind. That is the deliberate exception to the rule above.
+ */
+ recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError);
+ }
+ }
+ }
+ catch (Throwable recoveryFailure)
+ {
+ /*
+ * The give-back is the road which hands the change over, so a throw out of it - an
+ * allocation which fails in its turn, where what unwound the replay was a JVM out
+ * of memory - would leave the change owned by this thread after all. Hand it back
+ * bare, without the failure count that road did not reach, and restart the session
+ * so that it is delivered again: this is the last resort, the throwable is rethrown
+ * whatever happens here, and the thread this runs on may well be ending on it - so
+ * a request left for the next recovery of this domain to pick up is a request which
+ * may never be run.
+ */
+ if (owned != null)
+ {
+ remotePendingChanges.replayFailed(owned);
+ sessionRestartRequested.set(true);
+ /*
+ * Reported and restarted under guards of their own, and in that order: the report
+ * is the line an operator acts on, and the restart is what has the change
+ * delivered again - so a report which can not be formatted, on a road an
+ * OutOfMemoryError leads to, must not cost the restart.
+ */
+ try
+ {
+ logger.error(ERR_REPLAY_GIVE_BACK_FAILED, owned, getBaseDN(),
+ stackTraceToSingleLineString(recoveryFailure));
+ }
+ catch (Throwable reportFailure)
+ {
+ suppress(recoveryFailure, reportFailure);
+ }
+ try
+ {
+ runRequestedSessionRestarts(false);
+ }
+ catch (Throwable restartFailure)
+ {
+ /*
+ * Nothing is left to try: the change is listed, uncommitted and unowned, so any
+ * later session restart of this domain delivers it again. This goes with the
+ * throwable which is rethrown below rather than being reported on its own.
+ */
+ suppress(recoveryFailure, restartFailure);
+ }
+ }
+ // The error which unwound the replay is the one reported, whatever the give-back
+ // ran into on top of it.
+ suppress(t, recoveryFailure);
+ }
+ throw t;
+ }
+ }
+
+ /**
+ * Records the second throwable as one the first suppressed, unless the two are one and
+ * the same.
+ * <p>
+ * A JVM which has run out of memory hands out the error it prepared before it ran out as
+ * often as it is asked for one, so the roads out of a replay can carry the same instance
+ * twice - and a throwable can not suppress itself.
+ *
+ * @param thrown the throwable which is reported
+ * @param alsoThrown what was met on the way out of it
+ */
+ private static void suppress(Throwable thrown, Throwable alsoThrown)
+ {
+ if (thrown != alsoThrown)
+ {
+ thrown.addSuppressed(alsoThrown);
+ }
+ }
+
+ /**
+ * Replays the change of the provided message, then the changes which were waiting for
+ * it, for as long as there are some.
+ *
+ * @param msg
+ * The UpdateMsg to be replayed.
+ * @param replayThreadShutdown
+ * whether the replay thread was asked to stop
+ */
+ private void replayChangeAndTheChangesWaitingForIt(
+ LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown)
+ {
// Try replay the operation, then flush (replaying) any pending operation
// whose dependency has been replayed until no more left.
do
@@ -2582,6 +2745,13 @@
boolean replayAbandoned = false;
String replayErrorMsg = null;
CSN csn = null;
+ /*
+ * Read once, before anything of this delivery has run, so that the report of an ack
+ * which could not be published names the change even when reading it off the message
+ * is what threw: that report is written on the way out of a road which is already
+ * failing, and it must not be the throw which unwinds the replay.
+ */
+ final CSN delivered = msg.getCSN();
try
{
// The next operation for which to attempt replay.
@@ -2922,7 +3092,57 @@
replayErrorMsg = message.toString();
replayFailed = true;
}
- } finally
+ }
+ catch (OutOfMemoryError e)
+ {
+ /*
+ * The JVM is out of memory, which is not something to carry on replaying from: the
+ * error is left to unwind the replay thread, which ends on it (issue #923). It is
+ * not turned into a report of its own here either - the stack trace of an error
+ * which can be raised anywhere says little, and the uncaught exception handler of
+ * DirectoryThread writes the one line this is worth, with an alert.
+ *
+ * The other errors of the JVM take the road below. A StackOverflowError is met by
+ * the thread which recursed and is gone once the stack has unwound, and the entry
+ * being replayed is what raises it rather than the state of this server: ending a
+ * thread on it would have one change this replica can not replay cost it a replay
+ * thread per delivery, and nothing creates a replay thread to replace one which
+ * ends.
+ *
+ * The change is given back, counted as failing and asked for again on the way out
+ * (issue #922). The one thing done here is to make the ack this delivery publishes
+ * below say that the change was not applied: a replica which is asking for a change
+ * again must not have told an assured write that it is in the data here. It is a
+ * constant rather than a message built from the change, because building one asks
+ * the JVM for the memory it has just refused.
+ */
+ replayErrorMsg = REPLAY_RAN_OUT_OF_MEMORY;
+ throw e;
+ }
+ catch (Error e)
+ {
+ /*
+ * An Error out of the replay - a LinkageError met where a plugin or a backend class
+ * is loaded, an AssertionError - unwinds every road this replay has out of here, the
+ * ones which give the change back among them. The change is not in the data, so it
+ * is reported and given back here, on the road every other failed replay takes: the
+ * ack below says it was not applied, the failure counts against the give-up budget
+ * of the change, the session is restarted for it to be delivered again (issue #922)
+ * and the changes which were waiting for this one are replayed rather than left
+ * waiting for a thread to hand them out.
+ *
+ * It is not given up on where no operation could be built from the message, which is
+ * what an Exception at that point means: an Error says that this server could not run
+ * the replay, not that the message is one no delivery could ever build an operation
+ * from.
+ */
+ final LocalizableMessage message = ERR_ERROR_REPLAYING_CHANGE.get(
+ msg.getCSN(), getBaseDN(), stackTraceToSingleLineString(e));
+ logger.error(message);
+ replayErrorMsg = message.toString();
+ replayFailed = true;
+ }
+ finally
{
if (!dependency)
{
@@ -2934,7 +3154,88 @@
* down, so nothing would reach the server which is waiting for it, and an
* assured write would wait out its timeout rather than be told what happened.
*/
- processUpdateDone(msg, replayErrorMsg);
+ try
+ {
+ processUpdateDone(msg, replayErrorMsg);
+ }
+ catch (OutOfMemoryError e)
+ {
+ /*
+ * The one throw from here which is not caught, on the same terms as the arm
+ * above: a JVM which has run out of memory is not something to carry on
+ * replaying from, the error is left to end this replay thread, and the uncaught
+ * exception handler of DirectoryThread raises the alert #923 is about. Swallowed
+ * here instead, it would have this thread go on to the roads below and to the
+ * next change of the loop on an exhausted heap, with nothing reported anywhere.
+ *
+ * What the give-back on the way out of replay() then finds depends on the road
+ * the replay took to get here. A replay which failed, or which was unwinding on
+ * 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.
+ */
+ throw e;
+ }
+ catch (Throwable ackFailure)
+ {
+ /*
+ * Publishing the ack says nothing about whether the change was applied, so a
+ * throw here must not be a road out of the replay. It would step over the
+ * give-back of the change and, where the change was committed, over the
+ * getNextUpdate() below - the one drain of the changes this thread parked as
+ * waiting for it - leaving them waiting for a thread which is not replaying
+ * anything anymore.
+ *
+ * A master which is waiting for the ack waits out its assured timeout either
+ * way - and there is one only for an assured write in safe-read mode, which is
+ * the one delivery a replica acknowledges. processUpdateDone() runs for every
+ * delivery all the same: it accounts for the delivery in the receive window and
+ * in the processed-updates counter, and a throw from there is a throw from this
+ * bookkeeping, with no ack owed to anybody. Either way it is reported and the
+ * replay carries on to the road the change itself decided: applied, failed and
+ * asked for again, or given up on.
+ *
+ * Every step of processUpdateDone() catches what it can meet - the broker keeps
+ * a failure to publish to itself and retries it - so what reaches here is what
+ * no code of the replication protocol expected: an Error, and the tests drive
+ * it as one.
+ */
+ try
+ {
+ logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(),
+ stackTraceToSingleLineString(ackFailure));
+ }
+ catch (Throwable reportFailure)
+ {
+ /*
+ * Guarded like the report of a give-back which failed: this one runs on the
+ * same kind of road - the stack of the throwable is walked to build the line -
+ * and a report which can not be built must not become the throw which unwinds
+ * the replay past the give-back and past getNextUpdate().
+ *
+ * The line is tried once more with the name of the error alone, which walks no
+ * stack. Nothing rethrows what was caught here, so an error recorded as
+ * suppressed on it would be recorded nowhere, and this is the road on which an
+ * operator has the least to go on. A second refusal leaves nothing to say it
+ * with, and the replay carries on with what the change decided.
+ */
+ try
+ {
+ logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(),
+ ackFailure.getClass().getName());
+ }
+ catch (Throwable secondReportFailure)
+ {
+ // Nothing is left to say it with.
+ }
+ }
+ }
}
}
@@ -3191,6 +3492,29 @@
*/
private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean replayThreadShutdown)
{
+ return recoverFromReplayFailure(csn, replayThreadShutdown, false);
+ }
+
+ /**
+ * Recovers from a change which could not be replayed, telling apart the replay which was
+ * unwound by a JVM out of memory.
+ *
+ * @param csn
+ * the CSN of the change which could not be replayed
+ * @param replayThreadShutdown
+ * whether the replay thread was asked to stop
+ * @param outOfMemory
+ * whether what unwound the replay was the JVM running out of memory, in which
+ * case the line which says the change is being asked for again is not built -
+ * the report of a change which is given up on is, since it is what says this
+ * replica has diverged - and the session is restarted without the backoff
+ * @return {@code true} when the caller must stop replaying because the session is
+ * being restarted or is going away, {@code false} when it may carry on with
+ * the changes which follow
+ */
+ private boolean recoverFromReplayFailure(
+ CSN csn, AtomicBoolean replayThreadShutdown, boolean outOfMemory)
+ {
/*
* The failure is recorded, and the change given up on, while this thread still owns
* it: a change which is listed, uncommitted and unowned is what putRemoteUpdate()
@@ -3254,7 +3578,16 @@
return true;
}
- logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
+ if (!outOfMemory)
+ {
+ /*
+ * Not on the road out of a JVM which has run out of memory: building this line asks
+ * it for the memory it has just refused, and the ack of the delivery already says
+ * that the change was not applied. The constant that ack carries exists for the same
+ * reason.
+ */
+ logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
+ }
/*
* This change is not owned by anyone anymore, so the session has to be restarted for
* the replication server to deliver it again. Ask for the restart before trying to
@@ -3267,9 +3600,12 @@
* A replay thread which is stopping - the number of them is being changed - restarts
* the session all the same: nothing else would ask for the change it just released,
* and the ServerState would stay behind it for good. It does not sit through the
- * backoff on its way out, though: the backend is not what is going away.
+ * backoff on its way out, though: the backend is not what is going away. Neither does
+ * the thread an OutOfMemoryError is ending, for the same reason - and the restart is
+ * run rather than left to be asked for again, because that thread will not be there to
+ * run it, and a change nobody asks for again holds this domain's ServerState back.
*/
- runRequestedSessionRestarts(!replayThreadShutdown.get());
+ runRequestedSessionRestarts(!replayThreadShutdown.get() && !outOfMemory);
return true;
}
@@ -3292,7 +3628,41 @@
{
while (sessionRestartRequested.getAndSet(false))
{
- restartSession(wait);
+ boolean restarted = false;
+ try
+ {
+ restartSession(wait);
+ restarted = true;
+ }
+ finally
+ {
+ if (!restarted)
+ {
+ /*
+ * The request is put back where it was taken from. The flag is read and
+ * cleared before the restart runs, so a restart which ends abruptly - the
+ * session is stopped first, and starting it again creates a listener thread,
+ * which the operating system can refuse - would otherwise leave this domain
+ * with no session and with nothing left to ask for one.
+ *
+ * What a request left standing buys is bounded, and the bound is worth
+ * stating. Its two readers are the roads out of a failed and of an abandoned
+ * replay of this domain, and with no listener thread nothing is delivered
+ * anymore: the replays left to run are the changes already taken off the
+ * session - the ones waiting in the replay queue, and the ones parked as
+ * dependencies. One of those failing finds the request standing and runs the
+ * restart, which starts from a clean state, since disableService() drops the
+ * listener thread which was never started. Once they are spent, the domain
+ * stays down until it is disabled and enabled back, or the server is
+ * restarted. That is said where it can be heard: a refused thread is an
+ * OutOfMemoryError, and one which leaves recoverFromReplayFailure() or
+ * abandonReplay() ends the replay thread it is met on, so the uncaught
+ * exception handler of DirectoryThread writes the line and raises the alert,
+ * with the start of the listener thread in the trace.
+ */
+ sessionRestartRequested.set(true);
+ }
+ }
}
}
finally
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
index 34f4c4f..fa96a19 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
@@ -17,6 +17,8 @@
*/
package org.opends.server.replication.plugin;
+import net.jcip.annotations.GuardedBy;
+
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -36,11 +38,18 @@
*/
private volatile UpdateMsg msg;
/**
- * Whether a replay thread owns this change: it is being replayed, or it waits for the
+ * The replay thread which owns this change: it is being replayed, or it waits for the
* change it depends on. A remote change which no thread owns is one whose replay
* failed and which the replication server is expected to deliver again.
+ * <p>
+ * The owner is kept rather than the bare fact that there is one, so that a change is
+ * given back by the thread it was handed to and by nobody else: a release which arrives
+ * from a thread which does not own the change anymore - it reports a failure on a change
+ * which has been taken over since - would hand a change which is being replayed right
+ * now to a second thread (issue #922).
*/
- private boolean owned;
+ @GuardedBy("RemotePendingChanges.pendingChangesLock")
+ private Thread owner;
/**
* How many times in a row the replay of this change failed, and when the first of
* those failures happened - on a clock which only moves forward.
@@ -138,18 +147,30 @@
*/
public boolean isOwned()
{
- return owned;
+ return owner != null;
}
/**
- * Sets whether a replay thread owns this change.
+ * Returns whether the provided thread owns this change.
*
- * @param owned {@code true} when a replay thread takes the change over, {@code false}
- * when its replay failed and the change must be delivered again
+ * @param thread the thread which claims the change
+ * @return {@code true} if that thread is the one this change was handed to
*/
- public void setOwned(boolean owned)
+ public boolean isOwnedBy(Thread thread)
{
- this.owned = owned;
+ // A change nobody owns is not owned by a caller which has no thread to name either.
+ return thread != null && owner == thread;
+ }
+
+ /**
+ * Sets the replay thread which owns this change.
+ *
+ * @param owner the thread which takes the change over, or {@code null} when it is given
+ * back - its replay failed, or it has been applied
+ */
+ public void setOwner(Thread owner)
+ {
+ this.owner = owner;
}
/**
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 1d54c51..ec3ef7b 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
@@ -23,6 +23,8 @@
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -73,6 +75,32 @@
*/
private final ConcurrentSkipListSet<PendingChange> activeAndDependentChanges = new ConcurrentSkipListSet<>();
+ /**
+ * The change each replay thread is replaying right now, read by the give-back on the way
+ * out of a replay which was unwound.
+ * <p>
+ * It is an index of what {@link PendingChange#isOwnedBy(Thread)} already says rather than
+ * a second copy of it: it is written in the same locked step wherever a change is taken
+ * over or given back, and every road which acts on what it answers checks the ownership
+ * of the change again. What it buys is the read. That read is made on a road an
+ * {@code OutOfMemoryError} leads to, and looking the change up by walking the pending
+ * changes takes two locks and allocates an iterator - an allocation a JVM which has just
+ * refused one may well refuse again, and the change would then be left listed,
+ * uncommitted and owned by a thread which is not replaying it anymore, which is the wedge
+ * 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.
+ * <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)}
+ * clears it under the read lock, where every other writer holds the write lock, and it
+ * races nothing for it. {@link #clear()} is the one writer of every entry, and it holds
+ * both locks.
+ */
+ private final ConcurrentMap<Thread, CSN> changeBeingReplayed = new ConcurrentHashMap<>();
+
private final ReentrantReadWriteLock pendingChangesLock = new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock.ReadLock pendingChangesReadLock = pendingChangesLock.readLock();
private final ReentrantReadWriteLock.WriteLock pendingChangesWriteLock = pendingChangesLock.writeLock();
@@ -209,9 +237,25 @@
/**
* Mark an update message as committed.
+ * <p>
+ * A change another replay thread owns is not this thread's to record: it is reported as
+ * a change which is not here, the way one which is not listed anymore is. A thread
+ * decides that a change is in the data, or that it is to be given up on, and records
+ * that decision a turn of this lock later - long enough for the change to have been
+ * handed back, delivered again and taken over in between. Recording it then would
+ * advance the ServerState over a change which is not in the data yet (issue #889) and
+ * have the thread which is applying it right now fail to commit.
+ * <p>
+ * The give-back on the way out of an unwound replay is what makes this reachable: it
+ * runs wherever the replay was left, so the checks which keep it from taking a change
+ * away from the thread which owns it now belong on every road which reads ownership,
+ * not only on {@link #replayFailed(CSN)} (issue #922).
*
* @param csn
* The CSN of the update message that must be set as committed.
+ * @throws NoSuchElementException
+ * if there is no change with that CSN for this thread to record: it is not
+ * listed as pending anymore, or another replay thread owns it
*/
public void commit(CSN csn)
{
@@ -219,12 +263,14 @@
try
{
PendingChange curChange = pendingChanges.get(csn);
- if (curChange == null)
+ if (curChange == null
+ || (curChange.isOwned() && !curChange.isOwnedBy(Thread.currentThread())))
{
throw new NoSuchElementException();
}
curChange.setCommitted(true);
- curChange.setOwned(false);
+ curChange.setOwner(null);
+ changeBeingReplayed.remove(Thread.currentThread(), csn);
activeAndDependentChanges.remove(curChange);
final Iterator<PendingChange> it = pendingChanges.values().iterator();
@@ -271,6 +317,12 @@
* The changes another replay thread is applying right now are left alone: they are
* about to commit, and forgetting them would have their {@code commit()} fail, the
* ServerState stay behind them and the replication server replay them a second time.
+ * <p>
+ * Only the thread which owns the change gives it back. A release which arrives from
+ * another one is a release of a change which has been taken over since - the failure of
+ * a delivery is reported after the change it carried was handed to the delivery which
+ * follows it - and taking the change away from the thread which is replaying it right
+ * now is the double replay the ownership is there to prevent (issue #922).
*
* @param csn the CSN of the change whose replay failed
*/
@@ -280,9 +332,10 @@
try
{
final PendingChange change = pendingChanges.get(csn);
- if (change != null && !change.isCommitted())
+ if (change != null && !change.isCommitted() && change.isOwnedBy(Thread.currentThread()))
{
- change.setOwned(false);
+ change.setOwner(null);
+ changeBeingReplayed.remove(Thread.currentThread(), csn);
}
}
finally
@@ -339,9 +392,13 @@
* the CSN of the change whose replay failed
* @param nowMs
* when it failed, on a clock which only moves forward
- * @return the failures of the change, or {@code null} when it is not listed as an
- * uncommitted change anymore, which happens when the domain was disabled while
- * it was being replayed: there is no change left here to give up on
+ * @return the failures of the change, or {@code null} when there is no change left here
+ * for this thread to give up on: it is not listed as an uncommitted change
+ * anymore, which happens when the domain was disabled while it was being
+ * replayed, or another replay thread owns it, which is that change having been
+ * delivered again and taken over while this thread was on its way to reporting
+ * on it. Spending the give-up budget of a change another thread is applying
+ * would have this replica skip a change which is being written (issue #922)
*/
public ReplayFailure recordReplayFailure(CSN csn, long nowMs)
{
@@ -349,7 +406,8 @@
try
{
final PendingChange change = pendingChanges.get(csn);
- if (change == null || change.isCommitted())
+ if (change == null || change.isCommitted()
+ || (change.isOwned() && !change.isOwnedBy(Thread.currentThread())))
{
return null;
}
@@ -434,6 +492,7 @@
pendingChanges.clear();
dependentChanges.clear();
activeAndDependentChanges.clear();
+ changeBeingReplayed.clear();
failingChanges = 0;
}
finally
@@ -465,8 +524,16 @@
{
return false;
}
- change.setOwned(true);
+ /*
+ * Listed as being replayed before it is owned, and not the other way round: the
+ * caller enters the replay - where the give-back on the way out lives - once this
+ * returns, so nothing which allocates must run between the owner being stamped and
+ * that. A change listed here without an owner is the state a failed replay leaves
+ * behind, and the dependency checks which read this set do not read the owner.
+ */
activeAndDependentChanges.add(change);
+ changeBeingReplayed.put(Thread.currentThread(), change.getCSN());
+ change.setOwner(Thread.currentThread());
return true;
}
finally
@@ -475,7 +542,40 @@
}
}
/**
+ * 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).
+ * <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,
+ * and it runs on a road an {@code OutOfMemoryError} leads to - a lookup which allocated
+ * could be refused in its turn, and a give-back which does not know which change to give
+ * back leaves it listed, uncommitted and owned by a thread which is not replaying it
+ * anymore, which is the wedge this issue is about.
+ * <p>
+ * An answer which is out of date is safe: every road which acts on it - {@code commit()},
+ * {@link #recordReplayFailure(CSN, long)} and {@link #replayFailed(CSN)} - checks the
+ * ownership of the change again under the write lock, and is a no-op for a change this
+ * thread does not own anymore.
+ *
+ * @return the CSN of the change this thread is replaying, or {@code null} when it does
+ * not own one anymore - the road its replay took gave it back, or it was applied
+ */
+ CSN getChangeOwnedByCurrentThread()
+ {
+ return changeBeingReplayed.get(Thread.currentThread());
+ }
+
+ /**
* 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
+ * replayed by whichever replay thread cleared the change it was waiting for rather than
+ * by the one which parked it, and a change is given back by the thread which owns it
+ * and by nobody else (issue #922).
*
* @return The LDAPUpdateMsg to be handled.
*/
@@ -485,22 +585,65 @@
dependentChangesLock.lock();
try
{
- if (!dependentChanges.isEmpty() && !pendingChanges.isEmpty())
+ if (!hasChangeToHandOut())
{
- PendingChange firstDependentChange = dependentChanges.first();
- if (pendingChanges.firstKey().isNewerThanOrEqualTo(firstDependentChange.getCSN()))
- {
- dependentChanges.remove(firstDependentChange);
- return firstDependentChange.getLDAPUpdateMsg();
- }
+ /*
+ * Nothing is waiting, or what waits is still held back by the changes before it.
+ * This is called at the end of every replay, by every replay thread, so the answer
+ * is looked for under the read lock: taking the write lock here would have a
+ * backlog of waiting changes serialize the replay of the changes which have none.
+ */
+ return null;
}
- return null;
}
finally
{
dependentChangesLock.unlock();
pendingChangesReadLock.unlock();
}
+
+ /*
+ * There is one to hand out, and handing it out writes its owner, which is written
+ * under the write lock as the rest of the state of a change is. It is looked for again
+ * under that lock: another replay thread may have been handed it in between.
+ */
+ pendingChangesWriteLock.lock();
+ dependentChangesLock.lock();
+ try
+ {
+ if (hasChangeToHandOut())
+ {
+ final PendingChange firstDependentChange = dependentChanges.first();
+ /*
+ * Entered as the change this thread is replaying before it is taken out of the
+ * ones which are waiting, for the same reason markInProgress() enters it before it
+ * stamps the owner: an allocation which fails here must leave the change where it
+ * was rather than take it out of the hands which would hand it out again.
+ */
+ changeBeingReplayed.put(Thread.currentThread(), firstDependentChange.getCSN());
+ dependentChanges.remove(firstDependentChange);
+ firstDependentChange.setOwner(Thread.currentThread());
+ return firstDependentChange.getLDAPUpdateMsg();
+ }
+ return null;
+ }
+ finally
+ {
+ dependentChangesLock.unlock();
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
+ * Returns whether the first change waiting for another one can be replayed now, that is
+ * whether every change before it has left the pending changes.
+ */
+ @GuardedBy("pendingChangesLock, dependentChangesLock")
+ private boolean hasChangeToHandOut()
+ {
+ return !dependentChanges.isEmpty()
+ && !pendingChanges.isEmpty()
+ && pendingChanges.firstKey().isNewerThanOrEqualTo(dependentChanges.first().getCSN());
}
/**
@@ -524,6 +667,15 @@
{
dependentChanges.add(dependentChange);
}
+ /*
+ * Whichever of the two it was, this thread is not replaying that change anymore: a
+ * 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).
+ */
+ changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN());
}
finally
{
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 4a31d0c..0dddf3a 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
@@ -116,13 +116,33 @@
domain.replay(updateMsg, shutdown);
}
}
- catch (Exception e)
+ catch (OutOfMemoryError e)
+ {
+ /*
+ * The JVM is out of memory, which is not something to carry on replaying from: this
+ * thread does not stay for the changes which follow. Nothing is reported here - the
+ * uncaught exception handler of DirectoryThread is what says this thread is gone,
+ * with an alert - and the change it was replaying has been given back, counted and
+ * asked for again by the domain on its way out (issue #922).
+ *
+ * The other errors of the JVM are caught below: a StackOverflowError is gone once
+ * the stack has unwound, and a thread which ends here is one nothing replaces.
+ */
+ throw e;
+ }
+ catch (Throwable t)
{
/*
* catch all exceptions happening so that the thread never dies even
* in case of problems.
+ *
+ * An Error is not an Exception, so one raised here used to unwind run() and end
+ * this thread. Nothing creates a replay thread to replace it - the pool is created
+ * when the first domain of this server is - so the shared replay queue would have
+ * one consumer fewer for every domain, for as long as the server is up, until it
+ * has none left and replication stops (issue #923).
*/
- logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(e));
+ logger.error(ERR_EXCEPTION_REPLAYING_REPLICATION_MESSAGE, stackTraceToSingleLineString(t));
}
}
if (logger.isTraceEnabled())
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 7fa75ee..d5a7e69 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -657,6 +657,17 @@
to each other at the same time, of which one is dropped while the other one serves the domain. Until \
a handshake completes, no change is replicated over this connection, and the connection which \
completes one is reported in its turn
+ERR_ERROR_REPLAYING_CHANGE_315=An Error was thrown while replaying change %s in domain "%s": %s. \
+ The change has not been recorded as replayed and is given back to the replication server, which \
+ still owns it and sends it again
+ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain "%s" once its \
+ replay was done - that is where the delivery is accounted for and, for an assured write in \
+ safe-read mode, acknowledged: %s. Whether the change was applied is what its replay decided, \
+ and it is unaffected. A server which is waiting for that acknowledgement waits out its assured \
+ timeout instead
+ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \
+ server after the replay which owned it was unwound: %s. The change has been released without its \
+ failure being counted, and the session is being restarted so that the change is delivered again
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/plugins/ShortCircuitPlugin.java b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
index 20710c3..f61b23d 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
@@ -31,6 +31,7 @@
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -246,6 +247,9 @@
park.deregister();
}
parks.clear();
+ // Same shape: a throw which outlives the test which asked for it unwinds the replays of
+ // every test which follows, for as long as this plugin is loaded.
+ replayThrows.clear();
}
@@ -661,6 +665,23 @@
*/
if (operation.isSynchronizationOperation())
{
+ /*
+ * An error thrown here is thrown from inside the run() of the operation, which is
+ * where a plugin or a backend class which can not be loaded raises one: it unwinds
+ * the replay the way a real one does, rather than being reported as a result code the
+ * replay decides on. It comes before the park because it is the whole point of the
+ * delivery which asked for it, and no test asks for both on one operation.
+ */
+ final ThrownFromReplay thrower = replayThrows.get(key);
+ if (thrower != null)
+ {
+ final Error error = thrower.errorFor(operation);
+ if (error != null)
+ {
+ throw error;
+ }
+ }
+
final ParkedReplay park = parks.get(key);
if (park != null && park.parks(operation))
{
@@ -786,6 +807,82 @@
/** Registered parks for the replayed operations, keyed like the short circuits. */
private static final Map<String, ParkedReplay> parks = new ConcurrentHashMap<>();
+ /** The errors the replayed operations of one type throw, by operation type and section. */
+ private static final Map<String, ThrownFromReplay> replayThrows = new ConcurrentHashMap<>();
+
+ /**
+ * Throws an error out of the replay of the operations of one type, as many times as the
+ * test asked for.
+ * <p>
+ * The throw is made at a plugin point which runs inside {@code op.run()}, so it unwinds
+ * the replay from where a plugin or a backend class which can not be loaded raises one -
+ * past the point where the change was marked as being replayed by the thread which took
+ * it. That is what tells it apart from a delivery which reports a result code: a result
+ * code is a verdict the replay decided on, an error is the replay not running at all.
+ * <p>
+ * It is bounded rather than standing: the delivery which takes over from the one which
+ * was unwound has to be able to apply the change, or the test would watch this replica
+ * give up on a change it was never going to replay.
+ */
+ public static final class ThrownFromReplay
+ {
+ private final String key;
+ /** Which of the replayed operations of that type this throws out of. */
+ private final Predicate<PluginOperation> matches;
+ /** Built where it is thrown, so that it carries the stack of the replay it unwound. */
+ private final Supplier<? extends Error> error;
+ private final int maxTimes;
+ private final AtomicInteger thrown = new AtomicInteger();
+
+ private ThrownFromReplay(String key, Predicate<PluginOperation> matches,
+ Supplier<? extends Error> error, int maxTimes)
+ {
+ this.key = key;
+ this.matches = matches;
+ this.error = error;
+ this.maxTimes = maxTimes;
+ }
+
+ /**
+ * Returns the error to throw out of the provided operation, or {@code null} when this
+ * is not one of the operations it is for, or when it has been thrown as many times as
+ * the test asked for.
+ */
+ private Error errorFor(PluginOperation operation)
+ {
+ if (!matches.test(operation))
+ {
+ return null;
+ }
+ // Claimed before it is built, so that two replays of one change - the retry in place
+ // makes them - never take the same one twice.
+ if (thrown.incrementAndGet() > maxTimes)
+ {
+ return null;
+ }
+ return error.get();
+ }
+
+ /**
+ * Returns how many replays this threw out of.
+ *
+ * @return the number of replays which were unwound by this
+ */
+ public int thrownCount()
+ {
+ return Math.min(thrown.get(), maxTimes);
+ }
+
+ /**
+ * Stops throwing out of the replayed operations. A test must call this however it ends,
+ * or it leaves the deliveries of the tests which follow being unwound.
+ */
+ public void deregister()
+ {
+ replayThrows.remove(key, this);
+ }
+ }
+
/**
* Holds the replayed operations of one type where they are, one at a time, until the
* test lets each of them go.
@@ -1076,6 +1173,40 @@
return park;
}
+ /**
+ * Throws the provided error out of the replay of the operations of the given type, at the
+ * given plugin point, as many times as asked for and no more.
+ *
+ * @param operation the type of operation to throw out of
+ * @param section the plugin point to throw at, which can only be {@code PreParse}
+ * @param matches which of them to throw out of - the change a test acts on rather than
+ * whatever of that type reaches this point first
+ * @param error builds the error where it is thrown, so that it carries the stack trace of
+ * the replay it unwound
+ * @param maxTimes how many replays to unwind, after which the operations are let through:
+ * the delivery which takes over from the one which was unwound is what applies
+ * the change
+ * @return the throw, which the test must {@link ThrownFromReplay#deregister()} when it is
+ * done with it
+ * @throws IllegalArgumentException if asked for any plugin point but {@code PreParse}
+ */
+ public static ThrownFromReplay throwFromReplayedOperations(OperationType operation,
+ String section, Predicate<PluginOperation> matches, Supplier<? extends Error> error,
+ int maxTimes)
+ {
+ if (!"PreParse".equalsIgnoreCase(section))
+ {
+ // The pre-operation plugins are not invoked for synchronization operations at all, so
+ // a throw asked for anywhere else is one no replay would ever meet.
+ throw new IllegalArgumentException("replayed operations can only be thrown out of at"
+ + " PreParse, which is the only plugin point they reach, not at " + section);
+ }
+ final String key = keyFor(operation, section);
+ final ThrownFromReplay thrower = new ThrownFromReplay(key, matches, error, maxTimes);
+ replayThrows.put(key, thrower);
+ return thrower;
+ }
+
/** Returns the key a short circuit or a park of the given operations is kept under. */
private static String keyFor(OperationType operation, String section)
{
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 b2d03a7..b791c2a 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
@@ -33,9 +33,12 @@
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -58,6 +61,7 @@
import org.opends.server.plugins.PausePreParsePlugin;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.plugins.ShortCircuitPlugin.ParkedReplay;
+import org.opends.server.plugins.ShortCircuitPlugin.ThrownFromReplay;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
@@ -2584,13 +2588,738 @@
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed"));
+ assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseOperationRefusesAControl::new,
+ 18, "user.889.7", "the replay must fail once the operation is built");
+ }
+
+ /**
+ * The errors a replay meets which say nothing about the changes which follow: a class
+ * which could not be linked, and a stack which ran out on the entry being replayed. The
+ * second is an error of the JVM, but the thread which met it is whole again once the
+ * stack has unwound and the entry is what raised it, so ending the thread would have one
+ * change this replica can not replay cost it one replay thread per delivery.
+ * <p>
+ * Each row builds its error where it is thrown rather than here, so that the stack trace
+ * it carries is the one of the replay it unwound.
+ */
+ @DataProvider(name = "recoverableReplayErrors")
+ public Object[][] recoverableReplayErrors()
+ {
+ return new Object[][] {
+ { (Supplier<Error>) () -> new LinkageError("the replay of this change meets an Error"),
+ 19, "user.922.1", "a LinkageError" },
+ { (Supplier<Error>) StackOverflowError::new, 23, "user.922.5", "a StackOverflowError" },
+ };
+ }
+
+ /**
+ * Test case for [Issue 922] and [Issue 923]: a change whose replay threw an Error is
+ * given back, and the thread which was replaying it is still there to take the changes
+ * which follow.
+ * <p>
+ * A change is owned by the replay thread which took it, and that ownership is what
+ * keeps a change being replayed from being replayed a second time (OPENDJ-1115): every
+ * later delivery of it is refused as a duplicate. Ownership is given back on the roads
+ * which run to their end, so an Error - which unwinds the replay out of every one of
+ * them - would leave the change listed, uncommitted and owned by a thread which is not
+ * replaying it anymore: nobody could replay it, and this domain's ServerState would
+ * never move past it again.
+ * <p>
+ * The error is thrown where the operation is built, so what these rows pin is the arm
+ * which reports it and takes the road of a failed replay: it is caught inside the replay
+ * and never leaves it. The roads which do leave it - the give-back of a replay which was
+ * unwound, and the widened catch of the replay thread which is what keeps that thread
+ * alive - are pinned by
+ * {@link #aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain()}, where the throw is
+ * made past the point any catch of the replay runs on.
+ */
+ @Test(dataProvider = "recoverableReplayErrors")
+ public void aChangeWhoseReplayThrewAnErrorIsDeliveredAgain(
+ final Supplier<Error> error, int serverId, String uid, String what) throws Exception
+ {
+ testSetUp("aChangeWhoseReplayThrewAnErrorIsDeliveredAgain." + uid);
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseReplayThrewAnErrorIsDeliveredAgain " + uid));
+
+ assertChangeIsDeliveredAgainAfter(
+ (csn, dn, mods, entryUUID) ->
+ new ModifyMsgWhoseReplayThrows(csn, dn, mods, entryUUID, error),
+ serverId, uid, "the replay of this change throws " + what);
+ }
+
+ /**
+ * Test case for [Issue 922] and [Issue 923]: the change a replay thread was replaying
+ * when the JVM ran out of memory is given back before the error is left to end that
+ * thread.
+ * <p>
+ * An OutOfMemoryError is not turned into a failed replay and reported the way the other
+ * errors are - building the report asks for more of what the JVM has run out of - and
+ * the thread it unwinds is not replaced. The change it was replaying must not go with
+ * it: it is handed back so that the delivery which follows can replay it.
+ */
+ @Test
+ public void aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseReplayRanOutOfMemoryIsDeliveredAgain"));
+
+ /*
+ * The threads are read by identity rather than counted: what this test is about is the
+ * thread which met the error being gone, which is what #923 sanctions. Whether the pool
+ * is refilled afterwards is the half of that issue which is left open, and a count
+ * would freeze it here as the behaviour which is wanted.
+ */
+ final Set<Long> replayThreadsBefore = replayThreadIds();
+ final int initialUncaughtAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION);
+ assertChangeIsDeliveredAgainAfter(
+ (csn, dn, mods, entryUUID) -> new ModifyMsgWhoseReplayThrows(csn, dn, mods, entryUUID,
+ () -> new OutOfMemoryError("the JVM is out of memory")),
+ 20, "user.922.2", "the replay of this change meets an OutOfMemoryError");
+ /*
+ * The change is given back, and so replayed by another thread, while the thread which
+ * met the error is still unwinding - through the session restart, and then through the
+ * uncaught exception handler which raises the alert - so its end is waited for rather
+ * than read off the pool the moment the change lands. The alert is raised by that
+ * handler once run() has returned, so it lands after the thread is gone, and it is
+ * waited for in the same breath.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertFalse(replayThreadIds().containsAll(replayThreadsBefore),
+ "an OutOfMemoryError must end the replay thread which met it");
+ assertUncaughtExceptionAlertRaisedSince(initialUncaughtAlerts,
+ "the replay thread an OutOfMemoryError ended must have raised the alert #923 asks"
+ + " for on its way out");
+ }
+ });
+ }
+
+ /**
+ * Asserts that a {@code DirectoryThread} has ended on an uncaught throwable since the
+ * provided count of those alerts was read: the uncaught exception handler of its thread
+ * group is what raises it, and it is the one line a replay thread which an
+ * OutOfMemoryError ended leaves behind (issue #923).
+ * <p>
+ * At least one rather than exactly one: the alert is raised for every thread of the
+ * server which ends that way, and a thread of some other component ending during the
+ * test must not turn this into a failure of the wrong test.
+ */
+ private static void assertUncaughtExceptionAlertRaisedSince(int initialAlerts, String message)
+ {
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION))
+ .as(message)
+ .isGreaterThanOrEqualTo(initialAlerts + 1);
+ }
+
+ /**
+ * Test case for [Issue 922]: a change whose ack could not be published still takes the
+ * road its own replay decided.
+ * <p>
+ * The ack of a delivery is published in a finally which every road out of the replay runs
+ * through, and it is published on the session that delivery came over - which is being
+ * torn down when the replay of the change failed. Whether the change was applied is not
+ * something that publish can tell, so a throw there is reported and the replay carries on
+ * to the road the change itself decided: this one failed, so it is kept out of the
+ * ServerState, counted and asked for again.
+ * <p>
+ * Left to unwind, that throw would step over the give-back which follows it and leave the
+ * change owned by a thread which is not replaying it anymore.
+ */
+ @Test
+ public void aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseAckCouldNotBePublishedIsDeliveredAgain"));
+
+ assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseAckThrows::new,
+ 21, "user.922.3", "the ack of this change throws on the way out of the replay");
+ }
+
+ /**
+ * Test case for [Issue 922] and [Issue 923]: an OutOfMemoryError met where the ack of a
+ * delivery is published ends the replay thread, the way one met by the replay itself does.
+ * <p>
+ * A throw from the ack is caught so that it does not unwind the replay past the give-back
+ * of the change and past the hand-out of the changes which were waiting for it. A JVM
+ * which has run out of memory is the one exception to that: it is not something to carry
+ * on replaying from, so it is left to end this thread - the change is given back on the
+ * way out, and the uncaught exception handler of DirectoryThread writes the line and
+ * raises the alert #923 is about. Caught like every other throw from there, it would have
+ * this thread replay the changes which follow on an exhausted heap, with nothing said
+ * anywhere.
+ */
+ @Test
+ public void aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseAckRanOutOfMemoryIsDeliveredAgain"));
+
+ final Set<Long> replayThreadsBefore = replayThreadIds();
+ final int initialUncaughtAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_UNCAUGHT_EXCEPTION);
+ assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseAckRunsOutOfMemory::new,
+ 26, "user.922.10", "the ack of this change runs out of memory");
+ /*
+ * Waited for rather than read the moment the change lands: the change is given back -
+ * and so replayed by another thread - while the thread which met the error is still
+ * unwinding, through the session restart and then through the handler which raises the
+ * alert. The alert is what the rethrow is for, and the handler raises it once run() has
+ * returned, so it lands after the thread is gone and is waited for in the same breath.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertFalse(replayThreadIds().containsAll(replayThreadsBefore),
+ "an OutOfMemoryError met where the ack is published must end the replay thread"
+ + " which met it");
+ assertUncaughtExceptionAlertRaisedSince(initialUncaughtAlerts,
+ "the replay thread an OutOfMemoryError met where the ack is published ended must"
+ + " have raised the alert #923 asks for on its way out");
+ }
+ });
+ }
+
+ /**
+ * Test case for [Issue 922] and [Issue 923]: a change whose replay is unwound once the
+ * ack of its delivery is out is given back, and the thread which was replaying it stays.
+ * <p>
+ * The catches of the replay itself span the roads which decide what became of the change,
+ * and the ack is published once that is decided. What the replay runs afterwards - the
+ * give-back of a change which failed, and the hand-out of the changes which were waiting
+ * for it - is past every one of them: a throw there unwinds {@code replay()} with the
+ * change still owned by this thread, and a change owned by a thread which is not replaying
+ * it anymore is refused as a duplicate on every later delivery. So it is given back on the
+ * way out, and the error is left to the replay thread, whose catch is what keeps it alive
+ * for the changes which follow (issue #923).
+ */
+ @Test
+ public void aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseReplayIsUnwoundAfterItsAckIsDeliveredAgain"));
+
+ final Set<Long> replayThreadsBefore = replayThreadIds();
+ assertChangeIsDeliveredAgainAfter(ModifyMsgWhoseReplayIsUnwoundAfterItsAck::new,
+ 24, "user.922.6", "the replay of this change is unwound once its ack is out");
+ Assertions.assertThat(replayThreadIds())
+ .as("an Error which unwinds a replay must not end the thread which met it (issue #923)")
+ .containsAll(replayThreadsBefore);
+ }
+
+ /**
+ * Test case for [Issue 922]: a change whose replay keeps being unwound after its ack is
+ * given up on rather than asked for forever.
+ * <p>
+ * The change is given back on the road a failed replay takes rather than handed back bare,
+ * so the failure counts against the give-up budget of the change: a change which keeps
+ * unwinding the replays it is given to is eventually recorded as one this replica could
+ * not apply, and the administrator is told that it now diverges. Handed back bare it would
+ * be asked for, and have this domain restart its session for it, for as long as the server
+ * is up - which is the wedge this issue is about wearing another face.
+ */
+ @Test
+ public void aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn() throws Exception
+ {
+ testSetUp("aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseReplayKeepsBeingUnwoundAfterItsAckIsGivenUpOn"));
+
Entry tmp = TestCaseUtils.addEntry(
- "dn: uid=user.889.7," + baseDN,
+ "dn: uid=user.922.7," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
- "uid: user.889.7",
+ "uid: user.922.7",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final DN dn = tmp.getName();
+ final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+ setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
+ try
+ {
+ final CSN csn = new CSNGenerator(25, TimeThread.getTime()).newCSN();
+ final List<Modification> mods =
+ generatemods("description", "the replay of this change is unwound after its ack");
+
+ /*
+ * Nothing sends this change again - it never travelled a session - so every delivery
+ * of it is made here, until this replica gives up on the change whose replay it can
+ * not run to its end.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(csn))
+ {
+ domain.processUpdate(new ModifyMsgWhoseReplayIsUnwoundAfterItsAck(csn, dn, mods, uuid));
+ }
+ assertTrue(domain.getServerState().cover(csn),
+ "a change whose replay keeps being unwound must be given up on");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "the change which was given up on must be counted as failed, once");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ resetReplayGiveUpDelay();
+ }
+ }
+
+ /**
+ * Test case for [Issue 922]: the changes parked behind a change whose ack could not be
+ * published are replayed rather than left waiting for a thread which is gone.
+ * <p>
+ * A change which is waiting for another one is handed out by {@code getNextUpdate()},
+ * which the replay runs once it is done with the change it was given - after the ack of
+ * that delivery has been published. A throw from that ack used to unwind the replay past
+ * it, and the change had been committed by then, so nothing was owed back and nothing was
+ * handed out: the changes parked behind it stayed parked, and the ServerState of this
+ * domain stayed behind them until some other change was replayed here.
+ * <p>
+ * What tells the two apart is which thread replays the parked change. It is handed to
+ * whichever thread cleared the change it was waiting for, so it is replayed by the very
+ * thread which has just committed the change whose ack threw - a change nobody handed out
+ * is replayed by no one at all, and the wait below is what says so.
+ */
+ @Test
+ public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
+ {
+ testSetUp("theChangesParkedBehindAChangeWhoseAckFailedAreReplayed");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : theChangesParkedBehindAChangeWhoseAckFailedAreReplayed"));
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
+ final String parentUUID = "26262626-2626-2626-2626-262626262626";
+ final String childUUID = "27272727-2727-2727-2727-272727272727";
+ final Entry parent = TestCaseUtils.makeEntry(
+ "dn: ou=parked.922," + baseDN,
+ "objectClass: top",
+ "objectClass: organizationalUnit",
+ "ou: parked.922",
+ "entryUUID: " + parentUUID);
+ final Entry child = TestCaseUtils.makeEntry(
+ "dn: uid=user.922.8,ou=parked.922," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.922.8",
+ "cn: Aaccf Amar",
+ "sn: Amar",
+ "entryUUID: " + childUUID);
+
+ /*
+ * Both adds are held at the pre-parse plugin point, one at a time. The first park is
+ * what keeps the parent listed as pending - and owned by the thread replaying it - while
+ * the child is checked for dependencies, so the child is parked behind a change which is
+ * in flight rather than behind one which has already been applied.
+ */
+ final CSN parentCsn = gen.newCSN();
+ final CSN childCsn = gen.newCSN();
+ final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
+ OperationType.ADD, "PreParse",
+ op -> parentCsn.equals(OperationContext.getCSN(op))
+ || childCsn.equals(OperationContext.getCSN(op)));
+ try
+ {
+ domain.processUpdate(new AddMsgWhoseAckThrows(parentCsn, parent.getName(), parentUUID,
+ baseUUID, parent.getObjectClassAttribute(), parent.getAllAttributes()));
+ final Thread replayingParent = parked.awaitParked(60, SECONDS);
+
+ domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
+ child.getObjectClassAttribute(), child.getAllAttributes(), null));
+
+ /*
+ * The parent is applied and its ack throws where it is published. The replay carries
+ * on all the same, and the child is the change it hands itself next.
+ */
+ parked.release();
+ final Thread replayingChild = parked.awaitParked(60, SECONDS);
+ Assertions.assertThat(replayingChild)
+ .as("the change which was parked must be replayed by the thread which cleared what"
+ + " it was waiting for, rather than be left waiting")
+ .isSameAs(replayingParent);
+ parked.release();
+
+ assertNotNull(getEntry(child.getName(), 30000, true),
+ "the change which was parked behind the one whose ack threw must be applied");
+ }
+ finally
+ {
+ parked.deregister();
+ }
+ }
+
+ /**
+ * Test case for [Issue 922]: a change handed out as a dependency is given back when the
+ * replay it was handed to is unwound.
+ * <p>
+ * A change which was parked behind another one is handed out by {@code getNextUpdate()}
+ * to the thread which cleared what it was waiting for, and that thread owns it from then
+ * on. The give-back on the way out of an unwound replay asks which change this thread
+ * owns, so the hand-out has to be recorded where that question is answered, not only on
+ * the change: left out, the change would stay owned by a thread which is not replaying
+ * it anymore, and every later delivery of it would be refused as a duplicate - the wedge
+ * of this issue, on the dependency road.
+ * <p>
+ * The parent is held at the pre-parse plugin point while the child is delivered, so the
+ * child is parked behind a change in flight, and the thread which is thrown out of the
+ * child is read at the same plugin point: it must be the one which committed the parent,
+ * which is what says the child was handed out rather than taken off the queue. The child
+ * is thrown out of once, inside the replay, and then unwound past its ack, on the road
+ * every catch of the replay has already run on.
+ */
+ @Test
+ public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() throws Exception
+ {
+ testSetUp("aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound"));
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final CSNGenerator gen = new CSNGenerator(29, TimeThread.getTime());
+ final String parentUUID = "29292929-2929-2929-2929-292929292929";
+ final String childUUID = "30303030-3030-3030-3030-303030303030";
+ final Entry parent = TestCaseUtils.makeEntry(
+ "dn: ou=handed-out.922," + baseDN,
+ "objectClass: top",
+ "objectClass: organizationalUnit",
+ "ou: handed-out.922",
+ "entryUUID: " + parentUUID);
+ final Entry child = TestCaseUtils.makeEntry(
+ "dn: uid=user.922.11,ou=handed-out.922," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.922.11",
+ "cn: Aaccf Amar",
+ "sn: Amar",
+ "entryUUID: " + childUUID);
+
+ final CSN parentCsn = gen.newCSN();
+ final CSN childCsn = gen.newCSN();
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
+ OperationType.ADD, "PreParse", op -> parentCsn.equals(OperationContext.getCSN(op)));
+ /*
+ * The thread which reaches the plugin point with the child is the one replaying it, and
+ * it is read there rather than parked: a park would hold the child where the throw is
+ * made, and it is the throw which is wanted.
+ */
+ final AtomicReference<Thread> replayingChild = new AtomicReference<>();
+ final ThrownFromReplay thrown = ShortCircuitPlugin.throwFromReplayedOperations(
+ OperationType.ADD, "PreParse",
+ op ->
+ {
+ if (!childCsn.equals(OperationContext.getCSN(op)))
+ {
+ return false;
+ }
+ replayingChild.set(Thread.currentThread());
+ return true;
+ },
+ () -> new LinkageError("the replay of the change which was handed out meets an Error"),
+ 1);
+ try
+ {
+ domain.processUpdate(new AddMsg(parentCsn, parent.getName(), parentUUID, baseUUID,
+ parent.getObjectClassAttribute(), parent.getAllAttributes(), null));
+ final Thread replayingParent = parked.awaitParked(60, SECONDS);
+
+ domain.processUpdate(new AddMsgWhoseReplayIsUnwoundAfterItsAck(childCsn, child.getName(),
+ childUUID, parentUUID, child.getObjectClassAttribute(), child.getAllAttributes()));
+
+ // The parent is applied, and the child is the change its thread hands itself next.
+ parked.release();
+
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertEquals(thrown.thrownCount(), 1,
+ "the change which was handed out must have been thrown out of");
+ }
+ });
+ Assertions.assertThat(replayingChild.get())
+ .as("the change which was parked must be replayed by the thread which cleared what"
+ + " it was waiting for: that is the hand-out this test is about")
+ .isSameAs(replayingParent);
+
+ /*
+ * The child was thrown out of and its replay was then unwound, so it is not in the
+ * data and must not be in the ServerState - and it must not be given up on either: it
+ * is asked for again. The delivery which asks for it is made here, the way
+ * assertChangeIsDeliveredAgainAfter() makes it: nothing sends the change again, since
+ * it never travelled a session, and a delivery is dropped rather than queued while the
+ * session is being restarted. A delivery of a change a thread still owns is refused as
+ * the duplicate it is, which is where a change left owned by a thread which is not
+ * replaying it never comes back.
+ */
+ assertFalse(domain.getServerState().cover(childCsn),
+ "a change whose replay was unwound must be asked for again, not recorded as replayed");
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(childCsn))
+ {
+ domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
+ child.getObjectClassAttribute(), child.getAllAttributes(), null));
+ }
+ assertTrue(domain.getServerState().cover(childCsn),
+ "the change must be recorded as replayed once it has been delivered again");
+ }
+ });
+ assertNotNull(getEntry(child.getName(), 30000, true),
+ "the change which was handed out must be applied by the delivery which took over"
+ + " from the one which was unwound");
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
+ "a change which was delivered again must not be counted as one this replica gave up on");
+ assertTrue(replayingParent.isAlive(),
+ "an Error which unwinds a replay must not end the thread which met it (issue #923)");
+ }
+ finally
+ {
+ thrown.deregister();
+ parked.deregister();
+ }
+ }
+
+ /**
+ * Test case for [Issue 922]: the ack of a delivery whose replay threw an Error says that
+ * the change was not applied.
+ * <p>
+ * An assured write in SAFE_READ mode is told that its change is durable here by the ack
+ * this replica publishes, and it is published in a finally which every road out of the
+ * replay runs through - the roads an Error unwinds among them. A replica which is asking
+ * for a change again must never have told a master that the change is in its data, so the
+ * ack of a delivery whose replay threw reports the error and names this replica.
+ * <p>
+ * The error is thrown from a plugin point which runs inside the operation, so the change
+ * travels a real session and the ack can be read off the broker which published it - the
+ * counters can not report it, since handing the change back restarts the session and that
+ * resets every one of them.
+ */
+ @Test
+ public void theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt() throws Exception
+ {
+ testSetUp("theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : theAckOfADeliveryWhoseReplayThrewAnErrorReportsIt"));
+
+ final int serverId = 28;
+ /*
+ * In the group of the replication server, so that what is published below is one this
+ * domain has to acknowledge: an assured update from a broker of another group is
+ * acknowledged by the replication server itself, and says nothing about the replay.
+ */
+ ReplicationBroker broker =
+ openAssuredReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.922.9," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.922.9",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+
+ final CSN csn = gen.newCSN();
+ /*
+ * Thrown out of one replay and no more: the delivery which takes over from the one
+ * which was unwound is what applies the change, and a change this replica could never
+ * replay would be given up on rather than acknowledged twice.
+ */
+ final ThrownFromReplay thrown = ShortCircuitPlugin.throwFromReplayedOperations(
+ OperationType.DELETE, "PreParse",
+ op -> csn.equals(OperationContext.getCSN(op)),
+ () -> new LinkageError("the replay of this change meets an Error"), 1);
+ try
+ {
+ final DeleteMsg delete = new DeleteMsg(tmp.getName(), csn, uuid);
+ delete.setAssured(true);
+ delete.setAssuredMode(AssuredMode.SAFE_READ_MODE);
+ broker.publish(delete);
+
+ final AckMsg ack = awaitAck(broker, csn);
+ assertTrue(ack.hasReplayError(),
+ "the ack of a delivery whose replay threw an Error must report it rather than be"
+ + " the plain ack a master would take for a durable write");
+ assertFalse(ack.hasTimeout(),
+ "the ack must be the one the delivery published, not the one the replication"
+ + " server makes up when it gives up waiting for it");
+ Assertions.assertThat(ack.getFailedServers())
+ .as("the replica whose replay threw must be the one the ack names")
+ .containsExactly(domainSid);
+ assertEquals(thrown.thrownCount(), 1, "the replay of the change must have been unwound");
+ }
+ finally
+ {
+ thrown.deregister();
+ }
+
+ /*
+ * The change was given back rather than recorded as replayed, so the delivery which
+ * follows applies it - which is what the ack above said had not happened yet.
+ */
+ assertNull(getEntry(tmp.getName(), 30000, false),
+ "the change must be applied by the delivery which took over from the one whose"
+ + " replay threw");
+ assertTrue(domain.getServerState().cover(csn),
+ "the change must be recorded as replayed once it has been applied");
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 922]: a change whose replay keeps throwing an Error is given up
+ * on rather than asked for forever.
+ * <p>
+ * An Error takes the road every other failed replay takes, so the failures it leaves
+ * behind count against the give-up budget of the change: a change this replica can never
+ * apply must not hold its ServerState - and every change which follows it, from every
+ * master - back for good, whether its replay reported the failure or threw it.
+ */
+ @Test
+ public void aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn() throws Exception
+ {
+ testSetUp("aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseReplayKeepsThrowingAnErrorIsGivenUpOn"));
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.922.4," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.922.4",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final DN dn = tmp.getName();
+ final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+ setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
+ try
+ {
+ final CSN csn = new CSNGenerator(22, TimeThread.getTime()).newCSN();
+ final List<Modification> mods =
+ generatemods("description", "the replay of this change keeps throwing");
+
+ /*
+ * Nothing sends this change again - it never travelled a session - so every delivery
+ * of it is made here, until this replica gives up on the change it can not replay.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(csn))
+ {
+ domain.processUpdate(new ModifyMsgWhoseReplayThrows(csn, dn, mods, uuid,
+ () -> new LinkageError("the replay of this change meets an Error")));
+ }
+ assertTrue(domain.getServerState().cover(csn),
+ "a change whose replay keeps throwing must be given up on");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "the change which was given up on must be counted as failed, once");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ resetReplayGiveUpDelay();
+ }
+ }
+
+ /** A delivery of a change whose replay does not run to its end. */
+ private interface FailingDelivery
+ {
+ ModifyMsg newDelivery(CSN csn, DN dn, List<Modification> mods, String entryUUID);
+ }
+
+ /**
+ * Delivers a change whose replay does not run to its end, then checks that the change is
+ * neither recorded as replayed nor lost: the delivery which follows must be able to
+ * replay it.
+ *
+ * @param delivery the delivery whose replay is to be unwound
+ * @param serverId the replica the change comes from, one per test so that the CSNs of
+ * one are never covered by the ServerState another left behind
+ * @param uid the entry the change is made on
+ * @param description the value the change writes, once it is replayed
+ */
+ 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();
@@ -2601,12 +3330,10 @@
domain.resetUnreplayedChangeAlertThrottle();
final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
- final CSNGenerator gen = new CSNGenerator(18, TimeThread.getTime());
- final CSN csn = gen.newCSN();
- final String description = "the replay must fail once the operation is built";
+ final CSN csn = new CSNGenerator(serverId, TimeThread.getTime()).newCSN();
final List<Modification> mods = generatemods("description", description);
- domain.processUpdate(new ModifyMsgWhoseOperationRefusesAControl(csn, dn, mods, uuid));
+ domain.processUpdate(delivery.newDelivery(csn, dn, mods, uuid));
/*
* Long enough to outlast the session restart the failure asks for: a change which is
@@ -2615,24 +3342,22 @@
for (int i = 0; i < MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY; i++)
{
assertFalse(domain.getServerState().cover(csn),
- "a change whose operation was built must be asked for again, not recorded as replayed");
+ "a change whose replay was unwound must be asked for again, not recorded as replayed");
Thread.sleep(200);
}
assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
"a change which is still to be delivered again must not be counted as given up on");
- assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
- "a change which is still to be delivered again must not be alerted on as a divergence");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE),
+ initialAlerts, "a change which is still to be delivered again must not be alerted on"
+ + " as a divergence");
/*
- * The failed change is the barrier which holds this domain's ServerState back until
- * it is replayed, and the replication server sending it again is what replays it.
- * Nothing sends this one - it never travelled a session - so the delivery which takes
- * over from the one which failed is made here, and it is made until it is taken: a
- * delivery is dropped rather than queued while the listener thread is down, which it
- * is for as long as the recovery is restarting the session, and the monitor entry
- * read above comes back with the broker rather than with the listener. A delivery of
- * a change a replay thread owns is refused as the duplicate it is, and the ServerState
- * keeps this from delivering a change which was replayed a second time.
+ * Nothing sends this change again - it never travelled a session - so the delivery
+ * which takes over from the one which was unwound is made here, and it is made until
+ * it is taken: a delivery is dropped rather than queued while the listener thread is
+ * down, which it is for as long as the recovery is restarting the session. A delivery
+ * of a change a replay thread owns is refused as the duplicate it is, so this is where
+ * a change left owned by a thread which is gone never comes back.
*/
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
@@ -2652,7 +3377,83 @@
}
});
checkEntryHasAttributeValue(dn, "description", description, 30,
- "the change must be applied by the delivery which took over from the failed one");
+ "the change must be applied by the delivery which took over from the one which failed");
+ }
+
+ /**
+ * Returns the identities of the replay threads which are running.
+ * <p>
+ * Read by identity rather than counted where a test is about one thread in particular
+ * having ended: the pool belongs to the server rather than to a test, so a count says
+ * whether it is the size it was, not whether the thread which met the error is the one
+ * which is gone.
+ */
+ private static Set<Long> replayThreadIds()
+ {
+ final Set<Long> running = new HashSet<>();
+ for (Thread thread : Thread.getAllStackTraces().keySet())
+ {
+ if (thread.isAlive() && thread.getName().startsWith("Replica replay thread "))
+ {
+ running.add(thread.getId());
+ }
+ }
+ return running;
+ }
+
+ /**
+ * A ModifyMsg whose replay throws an Error.
+ * <p>
+ * It is thrown where the operation is built, which is inside the replay and past the
+ * point where the change was marked as being replayed by the thread which took it: what
+ * this pins is the road out of a replay which no {@code catch} of the replay itself used
+ * to run on.
+ */
+ private static final class ModifyMsgWhoseReplayThrows extends ModifyMsg
+ {
+ private final Supplier<Error> error;
+
+ private ModifyMsgWhoseReplayThrows(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID, Supplier<Error> error)
+ {
+ super(csn, dn, mods, entryUUID);
+ this.error = error;
+ }
+
+ @Override
+ public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
+ {
+ throw error.get();
+ }
+ }
+
+ /**
+ * A ModifyMsg whose replay fails and whose ack throws on the way out of it.
+ * <p>
+ * Its operation is built and can not be prepared for its replay, the way
+ * {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
+ * change owned by the replay thread. The ack of the delivery is then published in the
+ * finally every road out of the replay runs through, and this one throws there - which is
+ * what a session being torn down does.
+ */
+ private static final class ModifyMsgWhoseAckThrows
+ extends ModifyMsgWhoseOperationRefusesAControl
+ {
+ private ModifyMsgWhoseAckThrows(CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ /*
+ * An Error rather than the exception a session being torn down raises: the two take
+ * the same road, and an Error is what a catch of Exception would let past - the guard
+ * around the ack has to hold whatever publishing it threw.
+ */
+ throw new LinkageError("the ack of this delivery can not be published");
+ }
}
/**
@@ -2725,6 +3526,96 @@
}
/**
+ * A ModifyMsg whose ack runs out of memory on the way out of a replay which failed.
+ * <p>
+ * The replay fails first - its operation can not be prepared for the replay, the way
+ * {@code ModifyMsgWhoseOperationRefusesAControl} has it fail - so the change is one this
+ * replica asks for again, and the ack which says so is where the JVM runs out of memory.
+ * That is the one throw from there which is not caught: it ends the replay thread, and
+ * the change is given back on the way out.
+ */
+ private static final class ModifyMsgWhoseAckRunsOutOfMemory
+ extends ModifyMsgWhoseOperationRefusesAControl
+ {
+ private ModifyMsgWhoseAckRunsOutOfMemory(
+ 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 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
+ * road on which nothing is owed back to the replication server - and on which the changes
+ * parked behind this one are still waiting to be handed out.
+ */
+ private static final class AddMsgWhoseAckThrows extends AddMsg
+ {
+ private AddMsgWhoseAckThrows(CSN csn, DN dn, String entryUUID, String parentEntryUUID,
+ Attribute objectClasses, Iterable<Attribute> userAttributes)
+ {
+ super(csn, dn, entryUUID, parentEntryUUID, objectClasses, userAttributes, null);
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ // Read first thing by processUpdateDone(), which is what publishes the ack.
+ throw new LinkageError("the ack of this delivery can not be published");
+ }
+ }
+
+ /**
+ * An AddMsg whose replay is unwound once the ack of its delivery is out, the way
+ * {@link ModifyMsgWhoseReplayIsUnwoundAfterItsAck} is.
+ * <p>
+ * What has its replay fail is not the message but the test which delivers it, through a
+ * throw at the pre-parse plugin point: an add which is parked behind its parent has to be
+ * one whose operation builds and runs, or it would be given up on where no operation could
+ * be built from it. The throw is caught inside the replay, so it is what the replay runs
+ * once the ack is out - the give-back of the failed change - which reads the CSN off this
+ * message and is unwound by it.
+ */
+ private static final class AddMsgWhoseReplayIsUnwoundAfterItsAck extends AddMsg
+ {
+ private volatile boolean ackPublished;
+
+ private AddMsgWhoseReplayIsUnwoundAfterItsAck(CSN csn, DN dn, String entryUUID,
+ String parentEntryUUID, Attribute objectClasses, Iterable<Attribute> userAttributes)
+ {
+ super(csn, dn, entryUUID, parentEntryUUID, objectClasses, userAttributes, null);
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ // Read first thing by processUpdateDone(), and by nothing on the way in: a message
+ // handed to the domain rather than published is not one this server acknowledges.
+ ackPublished = true;
+ return super.isAssured();
+ }
+
+ @Override
+ public CSN getCSN()
+ {
+ if (ackPublished)
+ {
+ throw new LinkageError("the replay of this change is unwound once its ack is out");
+ }
+ return super.getCSN();
+ }
+ }
+
+ /**
* Test case for [Issue 908]: a domain being disabled - for an LDIF import, a restore, or
* a backend being taken offline - must not save its ServerState while a replay thread is
* half way through applying one of its changes.
@@ -3057,6 +3948,52 @@
}
/**
+ * A ModifyMsg whose replay is unwound once the ack of its delivery is out.
+ * <p>
+ * Its operation is built and can not be prepared for its replay, the way
+ * {@code ModifyMsgWhoseOperationRefusesAControl} has it, so the replay fails with the
+ * change owned by the replay thread. The CSN of the change is then read again - by the
+ * road which gives it back and asks for it again - and it is that read which throws here:
+ * past the ack, past the finally it is published in, and past every catch the replay
+ * itself has. So the only thing left to give the change back is the road out of
+ * {@code replay()}.
+ */
+ private static final class ModifyMsgWhoseReplayIsUnwoundAfterItsAck
+ extends ModifyMsgWhoseOperationRefusesAControl
+ {
+ private volatile boolean ackPublished;
+
+ private ModifyMsgWhoseReplayIsUnwoundAfterItsAck(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public boolean isAssured()
+ {
+ /*
+ * processUpdateDone() reads this first and reads nothing else of a delivery which is
+ * not assured, so the ack of this one is out by the time it returns. Nothing on the
+ * way in reads it: a message handed to the domain rather than published is not one
+ * this server acknowledges to anybody.
+ */
+ ackPublished = true;
+ return super.isAssured();
+ }
+
+ @Override
+ public CSN getCSN()
+ {
+ if (ackPublished)
+ {
+ throw new LinkageError("the replay of this change is unwound once its ack is out");
+ }
+ return super.getCSN();
+ }
+ }
+
+ /**
* 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
@@ -3066,7 +4003,7 @@
* protocol: {@code ModifyMsg.createOperation()} builds an operation whose controls can
* be added to, so this one is handed to the domain rather than published.
*/
- private static final class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
+ private static class ModifyMsgWhoseOperationRefusesAControl extends ModifyMsg
{
private ModifyMsgWhoseOperationRefusesAControl(
CSN csn, DN dn, List<Modification> mods, String entryUUID)
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 5f03553..e91292f 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
@@ -17,6 +17,11 @@
import static org.testng.Assert.*;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.atomic.AtomicReference;
+
import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.TestCaseUtils;
@@ -24,6 +29,8 @@
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyDNMsg;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -469,6 +476,371 @@
"a disabled domain forgot the change, so nothing is failing here anymore");
}
+ /**
+ * A change is given back by the replay thread which owns it and by nobody else.
+ * <p>
+ * The release is what lets the next delivery of a change be replayed, so a release
+ * which arrives from a thread which does not own the change hands a change which is
+ * being replayed right now to a second thread - the double replay the ownership is
+ * there to prevent (OPENDJ-1115). It happens when a thread reports a failure on a
+ * change which has been taken over since, which is every road out of a replay that is
+ * not the one which failed: an Error unwinding the replay thread, or an exception on
+ * the way to the ack (issue #922).
+ */
+ @Test
+ public void replayFailedIsIgnoredForAThreadWhichDoesNotOwnTheChange() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg delivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(delivery));
+ assertTrue(pendingChanges.markInProgress(delivery));
+
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ pendingChanges.replayFailed(csn);
+ }
+ });
+
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "a change another thread is replaying must not be taken over");
+ assertEquals(pendingChanges.getQueueSize(), 1);
+ assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), csn,
+ "a release which was ignored must leave the change with the thread which owns it");
+
+ // The thread which owns the change is still the one which decides its fate.
+ pendingChanges.replayFailed(csn);
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "the change its owner gave back must be taken over by the next delivery");
+ /*
+ * The give-back on the way out of an unwound replay reads this: a thread which gave a
+ * change back and is then unwound - the session restart it runs next can throw - must
+ * find nothing to give back, or it would count a second failure against a change it
+ * does not own anymore.
+ */
+ assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
+ "a change which was given back is not one this thread gives back again");
+ }
+
+ /**
+ * A change which was parked because it depends on another one is handed to whichever
+ * replay thread clears the change it was waiting for, rather than replayed by the
+ * thread which parked it. The thread it is handed to is the one which owns it from
+ * then on: the failure of the replay it is about to be given is reported by that
+ * thread, and a give-back which comes from a thread the change was never handed to is
+ * ignored (issue #922).
+ * <p>
+ * The hand-out is what the give-back on the way out of an unwound replay must read as
+ * well as the owner: a change handed out by {@code getNextUpdate()} whose replay is then
+ * unwound would otherwise be left owned by a thread which is not replaying it anymore,
+ * with every later delivery of it refused as a duplicate - the wedge of that issue, on
+ * the dependency road.
+ */
+ @Test
+ public void aChangeTakenAsADependencyIsOwnedByTheThreadWhichTakesIt() 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 rename must wait for the delete of the entry it renames into");
+
+ // The delete has been replayed, so the rename is handed to the thread which replayed it.
+ pendingChanges.commit(deleted);
+
+ final AtomicReference<LDAPUpdateMsg> taken = new AtomicReference<>();
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ taken.set(pendingChanges.getNextUpdate());
+ assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), renamed,
+ "the change handed out by getNextUpdate() must be the one the thread it was"
+ + " handed to gives back on the way out of an unwound replay");
+ // ... and the replay it was taken for failed.
+ pendingChanges.replayFailed(renamed);
+ assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
+ "a change which was given back is not one this thread gives back again");
+ }
+ });
+
+ assertSame(taken.get(), rename, "the change which was waiting must be handed out");
+ assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)),
+ "the change the thread which took it gave back must be taken over by the next delivery");
+ }
+
+ /**
+ * A thread which reports on a change it gave back a moment ago must not record it as
+ * replayed: the delivery which took the change over is being applied right now, so the
+ * ServerState would move past a change which is not in the data yet (issue #889), and
+ * the thread which is applying it would find nothing left to commit.
+ * <p>
+ * The give-back on the way out of an unwound replay is what makes this reachable: it
+ * runs wherever the replay was left, so the decision a thread carries and the record it
+ * makes of it can be a delivery apart (issue #922).
+ */
+ @Test
+ public void commitIsRefusedForAThreadWhichDoesNotOwnTheChange() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg delivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(delivery));
+ assertTrue(pendingChanges.markInProgress(delivery));
+
+ // The thread which gave up on an earlier delivery of this change records it as
+ // replayed while this delivery is being applied.
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ pendingChanges.commit(csn);
+ fail("a change another replay thread owns must not be recorded as replayed");
+ }
+ catch (NoSuchElementException expected)
+ {
+ // There is no change here for that thread to record, which is what its caller
+ // reports as ERR_OPERATION_NOT_FOUND_IN_PENDING.
+ }
+ }
+ });
+
+ assertTrue(state.isEmpty(),
+ "a change which is being applied must not be recorded in the ServerState");
+ assertEquals(pendingChanges.getQueueSize(), 1, "the change must stay listed as pending");
+
+ // The thread which owns the change records it once it really has been applied.
+ pendingChanges.commit(csn);
+
+ assertTrue(state.cover(csn));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+ }
+
+ /**
+ * The same holds for the failures which decide when this replica gives up on a change:
+ * a thread which does not own the change reports none, so the give-up budget of the
+ * delivery which took it over is left alone rather than spent by the one before it
+ * (issue #922).
+ */
+ @Test
+ public void recordReplayFailureIsIgnoredForAThreadWhichDoesNotOwnTheChange() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg delivery = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(delivery));
+ assertTrue(pendingChanges.markInProgress(delivery));
+
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ assertNull(pendingChanges.recordReplayFailure(csn, 1000),
+ "a change another replay thread owns is not this one's to give up on");
+ }
+ });
+
+ assertEquals(pendingChanges.recordReplayFailure(csn, 2000).getAttempts(), 1,
+ "only the failures of the delivery which owns the change must be counted");
+ }
+
+ /**
+ * 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).
+ * <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
+ * which was replaying it, and the change that thread is replaying afterwards is the next
+ * delivery it took - never a second one it holds at the same time.
+ */
+ @Test
+ public void theChangesParkedAsDependenciesAreNotOwnedByTheThreadWhichParkedThem()
+ 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 CSN taken = generator.newCSN();
+
+ /*
+ * The delete is being replayed by another thread of the pool, which is what a change
+ * waits for: the thread which parks a change is not the one applying the change it
+ * waits for.
+ */
+ final DeleteMsg delete = deleteMsg(deleted, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(delete));
+ runAndJoin(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ assertTrue(pendingChanges.markInProgress(delete),
+ "the delete must be listed as being replayed by the thread which took it");
+ }
+ });
+
+ // A rename into the DN that delete is on, parked by this very thread.
+ final ModifyDNMsg rename = renameIntoDeletedEntry(renamed);
+ assertTrue(pendingChanges.putRemoteUpdate(rename));
+ assertTrue(pendingChanges.markInProgress(rename));
+ assertTrue(pendingChanges.checkDependencies(rename),
+ "the rename must wait for the delete of the entry it renames into");
+
+ assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
+ "a change this thread parked as waiting for another one is not one it gives back:"
+ + " it is handed to whichever thread clears what it waits for");
+
+ // The delivery this thread took once the change it parked was out of its hands.
+ final DeleteMsg next = deleteMsg(taken, "uuid-3");
+ assertTrue(pendingChanges.putRemoteUpdate(next));
+ assertTrue(pendingChanges.markInProgress(next));
+
+ assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), taken,
+ "the change this thread is replaying is the one it must give back, not the one it"
+ + " parked as waiting for another");
+ }
+
+ /**
+ * A change which is in the data is not one to give back either, and it stays listed for as
+ * long as an older change holds the ServerState back: a give-back on the way out of an
+ * unwound replay must not ask for a change this replica has already applied (issue #922).
+ * <p>
+ * What this pins is that behaviour rather than one of the two conditions which hold it:
+ * {@code commit()} marks the change and clears its owner in the same write-locked step, so
+ * a committed change is never an owned one and the {@code isCommitted} arm of
+ * {@code getChangeOwnedByCurrentThread()} can not be told from the owner check by any test.
+ */
+ @Test
+ public void aChangeWhichWasAppliedIsNotOwnedAnymore() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN held = generator.newCSN();
+ final CSN applied = generator.newCSN();
+
+ // An older change nobody has replayed holds the ServerState back, so the change this
+ // thread applies stays listed here once it has been committed.
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(held, "uuid-1")));
+ final DeleteMsg delivery = deleteMsg(applied, "uuid-2");
+ assertTrue(pendingChanges.putRemoteUpdate(delivery));
+ assertTrue(pendingChanges.markInProgress(delivery));
+
+ assertEquals(pendingChanges.getChangeOwnedByCurrentThread(), applied,
+ "the change this thread is replaying must be the one it owns");
+
+ pendingChanges.commit(applied);
+
+ assertEquals(pendingChanges.getQueueSize(), 2,
+ "the change which was applied is held back by the one before it");
+ assertNull(pendingChanges.getChangeOwnedByCurrentThread(),
+ "a change which is in the data must not be given back on the way out of a replay");
+ }
+
+ /**
+ * A change which was waiting is handed out once: the thread it is handed to owns it from
+ * then on, and the threads which ask next are told there is nothing to take. Two threads
+ * handed the same change would replay it twice, which is what the ownership is there to
+ * prevent (OPENDJ-1115).
+ */
+ @Test
+ public void aChangeWhichWasWaitingIsHandedOutOnce() 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),
+ "the rename must wait for the delete of the entry it renames into");
+
+ assertNull(pendingChanges.getNextUpdate(),
+ "a change whose dependency still stands must not be handed out");
+
+ pendingChanges.commit(deleted);
+
+ assertSame(pendingChanges.getNextUpdate(), rename,
+ "the change which was waiting must be handed to the thread which cleared it");
+ assertNull(pendingChanges.getNextUpdate(),
+ "a change which has been handed out must not be handed out again");
+ }
+
+ /** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */
+ private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception
+ {
+ return new ModifyDNMsg(DN.valueOf("cn=uuid-2,dc=example,dc=com"), csn, "uuid-2",
+ null, false, null, "cn=uuid-1");
+ }
+
+ /**
+ * Runs the provided work on a thread of its own, waits for it and reports what it
+ * threw.
+ * <p>
+ * What the work throws has to be carried back here: an assertion which fails on another
+ * thread is lost to a bare {@link Thread#join()}, and every assertion these tests make
+ * about what a thread which does not own a change is answered is made on that thread.
+ */
+ private static void runAndJoin(Runnable runnable) throws Exception
+ {
+ final FutureTask<Void> task = new FutureTask<>(runnable, null);
+ final Thread thread = new Thread(task, "another replay thread");
+ thread.start();
+ thread.join();
+ try
+ {
+ task.get();
+ }
+ catch (ExecutionException e)
+ {
+ // Report what the work threw rather than the wrapper this task put around it: an
+ // assertion which failed is an Error, and it is the failure worth reading.
+ final Throwable cause = e.getCause();
+ if (cause instanceof Error)
+ {
+ throw (Error) cause;
+ }
+ if (cause instanceof Exception)
+ {
+ throw (Exception) cause;
+ }
+ throw e;
+ }
+ }
+
private DeleteMsg deleteMsg(CSN csn, String entryUUID) throws Exception
{
return new DeleteMsg(DN.valueOf("cn=" + entryUUID + ",dc=example,dc=com"), csn, entryUUID);
--
Gitblit v1.10.0