From d0422c684b5fc32280ec28813d2ab06a047ee63a Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 09 Sep 2026 07:02:11 +0000
Subject: [PATCH] [#909] Cover the change a stopped replay thread hands back to the replication server (#941)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 458 ++++++++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java | 351 +++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java | 55 +++
3 files changed, 856 insertions(+), 8 deletions(-)
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 751daac..20710c3 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
@@ -23,10 +23,14 @@
import java.io.IOException;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -230,6 +234,24 @@
/** {@inheritDoc} */
@Override
+ public void finalizePlugin()
+ {
+ /*
+ * A park which outlives the test which took it holds a replay thread of this server,
+ * and every replayed operation queued behind it, for as long as this plugin is
+ * loaded: the map is static and nothing but the test itself removes an entry from it.
+ */
+ for (ParkedReplay park : parks.values())
+ {
+ park.deregister();
+ }
+ parks.clear();
+ }
+
+
+
+ /** {@inheritDoc} */
+ @Override
public PluginResult.PreParse
doPreParse(PreParseAbandonOperation abandonOperation)
{
@@ -616,7 +638,7 @@
}
// Check for registered short circuits.
- final String key = operation.getOperationType() + "/" + section.toLowerCase();
+ final String key = keyFor(operation.getOperationType(), section);
Integer resultCode = shortCircuits.get(key);
if (resultCode != null)
{
@@ -630,6 +652,26 @@
// operations are let through, which is how a transient failure is simulated.
}
+ /*
+ * A parked replay is held here, which is inside the run() of the operation and before
+ * anything of the backend was taken: the thread which is replaying a change sits on
+ * this monitor while it still owns that change, which is what lets a test act on the
+ * thread rather than race it. It is consulted last, so that a park never takes an
+ * operation away from a control or from a registered short circuit.
+ */
+ if (operation.isSynchronizationOperation())
+ {
+ final ParkedReplay park = parks.get(key);
+ if (park != null && park.parks(operation))
+ {
+ final int parkResultCode = park.hold();
+ if (parkResultCode >= 0)
+ {
+ return parkResultCode;
+ }
+ }
+ }
+
// If we've gotten here, then we shouldn't short-circuit the operation
// processing.
return -1;
@@ -688,7 +730,7 @@
*/
public static int getShortCircuitCount(OperationType operation, String section)
{
- final AtomicInteger count = shortCircuitCounts.get(operation + "/" + section.toLowerCase());
+ final AtomicInteger count = shortCircuitCounts.get(keyFor(operation, section));
return count != null ? count.get() : 0;
}
@@ -701,7 +743,7 @@
*/
public static void registerShortCircuit(OperationType operation, String section, int resultCode)
{
- final String key = operation + "/" + section.toLowerCase();
+ final String key = keyFor(operation, section);
// This registration applies to every operation, and it counts from zero: a limit or
// a count left behind by a previous registration is not part of it.
shortCircuitCounts.remove(key);
@@ -720,7 +762,7 @@
*/
public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes)
{
- final String key = operation + "/" + section.toLowerCase();
+ final String key = keyFor(operation, section);
shortCircuitCounts.remove(key);
shortCircuitLimits.put(key, maxTimes);
shortCircuits.put(key, resultCode);
@@ -733,11 +775,310 @@
*/
public static void deregisterShortCircuit(OperationType operation, String section)
{
- final String key = operation + "/" + section.toLowerCase();
+ final String key = keyFor(operation, section);
shortCircuits.remove(key);
shortCircuitLimits.remove(key);
// The count belongs to the registration which is being removed: a test which counts
// the operations it short circuits must not inherit the count of the previous one.
shortCircuitCounts.remove(key);
}
+
+ /** Registered parks for the replayed operations, keyed like the short circuits. */
+ private static final Map<String, ParkedReplay> parks = new ConcurrentHashMap<>();
+
+ /**
+ * Holds the replayed operations of one type where they are, one at a time, until the
+ * test lets each of them go.
+ * <p>
+ * The hold is taken at a plugin point which runs inside {@code op.run()}, so the thread
+ * which is replaying a change is stopped while it still owns that change: a test can
+ * then do something to that thread - stop it, disable its domain - and know the change
+ * is in flight rather than hope it is. Nothing of the backend has been taken at that
+ * point, so a parked operation blocks the replay and nothing else.
+ */
+ public static final class ParkedReplay
+ {
+ /**
+ * The value which lets the operation run rather than short circuit it.
+ * <p>
+ * {@code ResultCode.UNDEFINED} is registered on {@code -1} as well, so
+ * {@code release(ResultCode.UNDEFINED.intValue())} lets the operation run instead of
+ * making it report that code - the same hole {@code registerShortCircuit(-1)} has.
+ * No caller has a use for it, and a park releases with a real result code or with
+ * none at all.
+ */
+ private static final int LET_THROUGH = -1;
+
+ /**
+ * How long an operation is held before this park gives up on the test which took it.
+ * <p>
+ * It is far longer than any release a test waits for - the fixture itself waits a
+ * minute for a park - and it exists for the test which never releases at all: a park
+ * leaked by a method killed on a timeout would otherwise hold a replay thread of this
+ * server, and every replayed operation queued behind it, for the life of the JVM.
+ */
+ private static final long MAX_HOLD_IN_MS = TimeUnit.MINUTES.toMillis(5);
+
+ private final String key;
+ /** Which of the replayed operations of that type this park is for. */
+ private final Predicate<PluginOperation> parked;
+ private final Object lock = new Object();
+ /** Whether an operation is parked right now. */
+ private boolean occupied;
+ /**
+ * The thread of the operation which parked last. It is never cleared, so that a test
+ * which waited for a park is handed the thread of that park even when the operation
+ * has left the park since - a park which is let go of by {@link #deregister()}, or by
+ * the thread it holds being interrupted, would otherwise hand out no thread at all
+ * and have an assertion on which thread replays the change pass without asserting it.
+ */
+ private Thread lastParkedThread;
+ /** How many operations were parked, which is what tells one park from the next. */
+ private int parkedOperations;
+ /** How many of them the test has waited for already. */
+ private int awaitedOperations;
+ private boolean released;
+ private int releasedResultCode;
+ private boolean deregistered;
+
+ private ParkedReplay(String key, Predicate<PluginOperation> parked)
+ {
+ this.key = key;
+ this.parked = parked;
+ }
+
+ /** Returns whether the provided operation is one this park is for. */
+ private boolean parks(PluginOperation operation)
+ {
+ return parked.test(operation);
+ }
+
+ /**
+ * Parks the calling operation until the test releases it. Runs on the thread which is
+ * replaying the change.
+ *
+ * @return the result code the operation must be short circuited with, or a negative
+ * value to let it run
+ */
+ private int hold()
+ {
+ final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(MAX_HOLD_IN_MS);
+ synchronized (lock)
+ {
+ // One operation at a time, so that a release belongs to the operation the test
+ // waited for rather than to whichever of them the scheduler let in first.
+ while (occupied && !deregistered)
+ {
+ if (!waitOnLock(deadline))
+ {
+ return LET_THROUGH;
+ }
+ }
+ if (deregistered)
+ {
+ return LET_THROUGH;
+ }
+ occupied = true;
+ lastParkedThread = Thread.currentThread();
+ parkedOperations++;
+ released = false;
+ lock.notifyAll();
+ try
+ {
+ while (!released && !deregistered)
+ {
+ if (!waitOnLock(deadline))
+ {
+ return LET_THROUGH;
+ }
+ }
+ return released ? releasedResultCode : LET_THROUGH;
+ }
+ finally
+ {
+ occupied = false;
+ lock.notifyAll();
+ }
+ }
+ }
+
+ /**
+ * Waits on the monitor until the provided deadline, reporting whether waiting can go
+ * on. A deadline which has passed gives up on this park altogether rather than only
+ * on the operation which reached it: the operations behind it would each pay the
+ * whole wait again otherwise.
+ */
+ private boolean waitOnLock(long deadlineInNanos)
+ {
+ final long leftInNanos = deadlineInNanos - System.nanoTime();
+ if (leftInNanos <= 0)
+ {
+ giveUpOnTheTest();
+ return false;
+ }
+ try
+ {
+ // Rounded up, so that a budget shorter than a millisecond is still waited out
+ // rather than truncated to a wait with no timeout at all.
+ lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
+ return true;
+ }
+ catch (InterruptedException e)
+ {
+ // Whatever wants this thread to stop wins over the park: let the operation run
+ // rather than hold a thread which is being taken down.
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ /** Stops parking anything and says so, after a test held an operation for too long. */
+ private void giveUpOnTheTest()
+ {
+ System.err.println("***** ERROR: a replayed operation was parked on " + key
+ + " for " + MAX_HOLD_IN_MS + " ms and was never released: the test which took"
+ + " this park left it behind. Letting the operation run and parking no more.");
+ deregister();
+ }
+
+ /**
+ * Waits for a replayed operation which was not waited for yet to be parked, and
+ * reports which thread is replaying it. The operations are parked one at a time, so
+ * that thread is the one which was parked when this returns; the thread of the last
+ * park is reported when several of them were let go of without being waited for.
+ *
+ * @param timeout how long to wait for it
+ * @param unit the unit of the timeout
+ * @return the thread which is replaying the parked operation
+ * @throws InterruptedException if this thread is interrupted while waiting
+ * @throws TimeoutException if no operation was parked in time
+ * @throws IllegalStateException if this park is gone, so that nothing can be parked
+ * on it any more
+ */
+ public Thread awaitParked(long timeout, TimeUnit unit)
+ throws InterruptedException, TimeoutException
+ {
+ final long deadline = System.nanoTime() + unit.toNanos(timeout);
+ synchronized (lock)
+ {
+ while (parkedOperations <= awaitedOperations)
+ {
+ if (deregistered)
+ {
+ // Waiting out the budget here would report a timeout naming the operations
+ // which never parked, rather than the park which cannot park them any more.
+ throw new IllegalStateException("the park on " + key + " is gone - it was"
+ + " deregistered, or displaced by another park of the same operations -"
+ + " so no replayed operation will be parked on it again");
+ }
+ final long leftInNanos = deadline - System.nanoTime();
+ if (leftInNanos <= 0)
+ {
+ throw new TimeoutException("no replayed operation was parked on " + key
+ + " within " + timeout + " " + unit);
+ }
+ // Rounded up, so that a budget shorter than a millisecond is still waited out
+ // rather than truncated to a wait with no timeout at all.
+ lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
+ }
+ awaitedOperations = parkedOperations;
+ return lastParkedThread;
+ }
+ }
+
+ /**
+ * Lets the parked operation run. Valid once {@link #awaitParked} has reported that
+ * operation: see there for what a release which arrives before it costs.
+ */
+ public void release()
+ {
+ release(LET_THROUGH);
+ }
+
+ /**
+ * Lets the parked operation go, short circuiting it with the provided result code.
+ * <p>
+ * Valid once {@link #awaitParked} has reported the operation being released. A
+ * release which arrives before an operation is parked is wiped by the park it was
+ * meant for - a park starts out unreleased - and that operation then waits for a
+ * release which has already been spent.
+ *
+ * @param resultCode the result code the operation must report
+ */
+ public void release(int resultCode)
+ {
+ synchronized (lock)
+ {
+ if (!occupied)
+ {
+ throw new IllegalStateException("nothing is parked on " + key + " to release:"
+ + " a release is spent by the park it arrives before, and the operation"
+ + " which parks next then waits for one which has already been given");
+ }
+ released = true;
+ releasedResultCode = resultCode;
+ lock.notifyAll();
+ }
+ }
+
+ /**
+ * Stops parking the replayed operations and lets go of the one which is parked, if
+ * any. A test must call this however it ends, or it leaves a replay thread of this
+ * server parked for good.
+ */
+ public void deregister()
+ {
+ parks.remove(key, this);
+ synchronized (lock)
+ {
+ deregistered = true;
+ lock.notifyAll();
+ }
+ }
+ }
+
+ /**
+ * Parks the replayed operations of the given type at the given plugin point, until the
+ * test releases each of them.
+ *
+ * @param operation the type of operation to park
+ * @param section the plugin point to park them at, which can only be {@code PreParse}
+ * @param parked which of them to park - the change a test acts on rather than whatever
+ * of that type reaches this point first, which is somebody else's change as
+ * soon as more than one of them is in flight
+ * @return the park, which the test must {@link ParkedReplay#deregister()} when it is
+ * done with it
+ * @throws IllegalArgumentException if asked for any plugin point but {@code PreParse}
+ */
+ public static ParkedReplay parkReplayedOperations(
+ OperationType operation, String section, Predicate<PluginOperation> parked)
+ {
+ if (!"PreParse".equalsIgnoreCase(section))
+ {
+ /*
+ * The pre-operation plugins are not invoked for synchronization operations at all,
+ * so a park anywhere else is never reached: the test which took it would wait out
+ * its whole budget for an operation which cannot park, and be told that none did
+ * rather than that none could.
+ */
+ throw new IllegalArgumentException("replayed operations can only be parked at"
+ + " PreParse, which is the only plugin point they reach, not at " + section);
+ }
+ final String key = keyFor(operation, section);
+ final ParkedReplay park = new ParkedReplay(key, parked);
+ final ParkedReplay previous = parks.put(key, park);
+ if (previous != null)
+ {
+ // A park a test left behind holds a replay thread of this server for good once the
+ // map stops pointing at it: let go of it rather than lose the last reference to it.
+ previous.deregister();
+ }
+ return park;
+ }
+
+ /** Returns the key a short circuit or a park of the given operations is kept under. */
+ private static String keyFor(OperationType operation, String section)
+ {
+ return operation + "/" + section.toLowerCase(Locale.ROOT);
+ }
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
index a1eeb60..bcedc42 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationTestCase.java
@@ -102,6 +102,16 @@
/** Generation id for a fully empty domain. */
public static final long EMPTY_DN_GENID = GenerationIdChecksum.EMPTY_BACKEND_GENERATION_ID;
+ /** The group a replication server and a replication domain are in unless told otherwise. */
+ protected static final int DEFAULT_GROUP_ID = 1;
+
+ /**
+ * The group of a broker which is in none. Assured replication does not cross group ids,
+ * so such a broker is never waited for and never waits: it is all a broker which only
+ * publishes and reads updates needs.
+ */
+ private static final int NO_GROUP_ID = -1;
+
/** How many times {@link #assertMonitorAttrValueStays} reads a value by default. */
private static final int MONITOR_ATTR_SAMPLES = 5;
@@ -241,7 +251,42 @@
int serverId, int windowSize, int port, int timeout,
long generationId) throws Exception
{
- final DomainFakeCfg config = newFakeCfg(baseDN, serverId, port);
+ return openReplicationSession(
+ newFakeCfg(baseDN, serverId, port), windowSize, timeout, generationId);
+ }
+
+ /**
+ * Open a session to the local ReplicationServer which takes part in assured replication.
+ * <p>
+ * Assured replication does not cross group ids, so a broker whose updates are to be
+ * acknowledged by the replicas of this server has to be in the group of the replication
+ * server: an update published by a broker of another group is acknowledged on the spot,
+ * by the replication server itself, and says nothing about what any replica did with it.
+ * <p>
+ * The group cuts both ways, and this broker does not acknowledge anything: the
+ * replication server expects an ack from every replica of its group whatever that
+ * replica is configured for, so a SAFE_READ update published by anyone else while this
+ * broker is connected waits out the {@code assured-timeout} of the server. Publish the
+ * assured updates from this broker, and open only one of them.
+ *
+ * @param baseDN the suffix the session is opened for
+ * @param serverId the id this broker takes
+ * @param windowSize the window size of the session
+ * @param port the port of the local replication server
+ * @param timeout the read timeout of the session, or 0 for none
+ * @return the connected broker
+ * @throws Exception if the session could not be opened
+ */
+ protected ReplicationBroker openAssuredReplicationSession(final DN baseDN,
+ int serverId, int windowSize, int port, int timeout) throws Exception
+ {
+ return openReplicationSession(newFakeCfg(baseDN, serverId, port, DEFAULT_GROUP_ID),
+ windowSize, timeout, getGenerationId(baseDN));
+ }
+
+ private ReplicationBroker openReplicationSession(final DomainFakeCfg config,
+ int windowSize, int timeout, long generationId) throws Exception
+ {
config.setWindowSize(windowSize);
final ReplicationBroker broker = new ReplicationBroker(
@@ -253,7 +298,13 @@
protected DomainFakeCfg newFakeCfg(final DN baseDN, int serverId, int port)
{
- DomainFakeCfg fakeCfg = new DomainFakeCfg(baseDN, serverId, newTreeSet("127.0.0.1:" + port));
+ return newFakeCfg(baseDN, serverId, port, NO_GROUP_ID);
+ }
+
+ protected DomainFakeCfg newFakeCfg(final DN baseDN, int serverId, int port, int groupId)
+ {
+ DomainFakeCfg fakeCfg =
+ new DomainFakeCfg(baseDN, serverId, newTreeSet("127.0.0.1:" + port), groupId);
fakeCfg.setHeartbeatInterval(100000);
fakeCfg.setChangetimeHeartbeatInterval(500);
return fakeCfg;
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 ba7056a..4a9df33 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
@@ -22,6 +22,7 @@
import static org.forgerock.opendj.ldap.ModificationType.*;
import static org.forgerock.opendj.ldap.requests.Requests.*;
import static org.forgerock.opendj.ldap.schema.CoreSchema.*;
+import static org.mockito.Mockito.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
@@ -32,6 +33,8 @@
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -44,6 +47,7 @@
import org.forgerock.opendj.ldap.requests.ModifyDNRequest;
import org.forgerock.opendj.ldap.requests.ModifyRequest;
import org.forgerock.opendj.ldap.schema.AttributeType;
+import org.forgerock.opendj.server.config.server.ReplicationSynchronizationProviderCfg;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.AddOperation;
import org.opends.server.core.DeleteOperation;
@@ -52,11 +56,14 @@
import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.extensions.DummyAlertHandler;
import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.plugins.ShortCircuitPlugin.ParkedReplay;
import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.plugin.LDAPReplicationDomain;
import org.opends.server.replication.plugin.MultimasterReplication;
+import org.opends.server.replication.protocol.AckMsg;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.HeartbeatThread;
@@ -157,7 +164,15 @@
+ "cn: Replication Server\n"
+ "ds-cfg-replication-port: " + replServerPort + "\n"
+ "ds-cfg-replication-db-directory: UpdateOperationTest\n"
- + "ds-cfg-replication-server-id: 107\n";
+ + "ds-cfg-replication-server-id: 107\n"
+ /*
+ * Long enough for a delivery which a test stops on its way through the replay:
+ * the acks of an assured update are waited for from the moment it is published,
+ * and the default second is spent long before a test which parks that delivery
+ * has let go of it. Nothing waits it out - no test here leaves an assured update
+ * unacknowledged - so it only bounds a failure.
+ */
+ + "ds-cfg-assured-timeout: 120000ms\n";
// suffix synchronized
String testName = "updateOperationTest";
@@ -1516,6 +1531,447 @@
}
/**
+ * Test case for [Issue 909]: a replay thread which is stopped while it holds a change -
+ * the number of replay threads is changed on a live server - must hand that change back
+ * to the replication server instead of leaving it listed as owned by a thread which is
+ * gone.
+ * <p>
+ * The change is parked inside the operation it is replayed by, so the thread is caught
+ * while it still owns it rather than raced for: a change which is released by the
+ * ordinary recovery instead ends the same way - delivered again and applied - so a test
+ * which only watched the end state would pass whether or not the hand-back happened.
+ * <p>
+ * What tells them apart is which thread replays the change next. The thread which held
+ * it is gone, and the change is replayed by one of the threads which replaced it, so
+ * the delivery it is replayed from can only be a new one: an attempt which the same
+ * thread made again would be the retry in place, and a change nobody handed back is
+ * never delivered again at all - it stays listed as owned by a thread which is gone,
+ * with the ServerState of this domain stopped behind it for good.
+ */
+ @Test
+ public void aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain() throws Exception
+ {
+ testSetUp("aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain"));
+
+ final int serverId = 14;
+ /*
+ * In the group of the replication server, so that the delete 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.909," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.909",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+ /*
+ * Only the counters of the replay are read: a session restart takes this domain
+ * through NOT_CONNECTED, which resets every monitoring counter of the replication
+ * service - the updates it received and processed, and the assured acks it sent -
+ * and handing a change back is a session restart.
+ */
+ final long initialApplied = getMonitorAttrValue(baseDN, "replayed-updates-ok");
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+
+ /*
+ * Hold the replayed deletes where they are. The park is taken at the pre-parse
+ * plugin point, which runs inside op.run() and before anything of the backend was
+ * taken: the replay thread stops there while it still owns the change, and the
+ * pre-operation plugins are not invoked for synchronization operations at all.
+ */
+ final CSN csn = gen.newCSN();
+ final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
+ OperationType.DELETE, "PreParse", op -> csn.equals(OperationContext.getCSN(op)));
+ final AtomicReference<Throwable> reconfigurationFailure = new AtomicReference<>();
+ Thread reconfiguration = null;
+ boolean reconfigurationFinished = true;
+ try
+ {
+ final DeleteMsg delete = new DeleteMsg(tmp.getName(), csn, uuid);
+ /*
+ * Published assured in SAFE_READ mode, so that the delivery which is abandoned
+ * has to say what it did. The ack is the one thing the counters cannot report
+ * afterwards - the session restart the hand-back performs resets every one of
+ * them - so it is read off this broker rather than counted.
+ */
+ delete.setAssured(true);
+ delete.setAssuredMode(AssuredMode.SAFE_READ_MODE);
+ broker.publish(delete);
+
+ // A replay thread now owns the change and is stopped inside its operation.
+ final Thread abandoningThread = parked.awaitParked(60, SECONDS);
+
+ /*
+ * Change the number of replay threads while that thread holds the change. It runs
+ * on a thread of its own because stopping the replay threads joins them: it can
+ * not return before the parked thread is let go, which is exactly the ordering
+ * this test is about.
+ */
+ reconfiguration = startReplayThreadReconfiguration(2, reconfigurationFailure);
+ awaitStoppingTheReplayThreads(reconfiguration, reconfigurationFailure);
+
+ /*
+ * The parked thread has been asked to stop by now, so its attempt comes back on a
+ * storage which did not serve the operation - which is retried in place - and the
+ * attempt which follows is where it finds out that it is going away and gives the
+ * change back instead.
+ */
+ parked.release(ResultCode.UNAVAILABLE.intValue());
+
+ /*
+ * The change was given back, so the replication server owns it again and delivers
+ * it once more. This second park is where the state of the domain is read: the
+ * change is held inside its operation, so nothing can be recording it while the
+ * assertions below run.
+ */
+ final Thread replayingThread =
+ awaitParkedOrReportReconfigurationFailure(parked, reconfigurationFailure);
+ reconfiguration.join(SECONDS.toMillis(60));
+ assertFalse(reconfiguration.isAlive(),
+ "the replay threads were reconfigured, but applyConfigurationChange never returned");
+ if (reconfigurationFailure.get() != null)
+ {
+ throw new AssertionError("the replay threads could not be reconfigured",
+ reconfigurationFailure.get());
+ }
+
+ /*
+ * The change is being replayed by another thread, and the thread which held it is
+ * gone: it gave the change back on its way out rather than take it with it. An
+ * attempt made by the thread which held it would be the retry in place instead,
+ * and a change which was never handed back is not replayed by anyone.
+ */
+ assertFalse(abandoningThread.isAlive(),
+ "the replay thread which held the change must have stopped");
+ Assertions.assertThat(replayingThread)
+ .as("the change must be replayed by a thread which did not hold it")
+ .isNotSameAs(abandoningThread);
+
+ assertFalse(domain.getServerState().cover(csn),
+ "a change a stopped replay thread never applied must not be in the ServerState");
+ assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
+ "a change which was handed back must not be counted as one this replica gave up on");
+ assertNotNull(getEntry(tmp.getName(), 1, true),
+ "the entry must not have been deleted by the delivery which was abandoned");
+
+ /*
+ * The delivery which was abandoned said what it did: an assured write which is
+ * waiting on this replica must not be told that the change is in the data here,
+ * because it is a change this replica is still asking for.
+ *
+ * What is read is the ack, not when it was published. The hand-back gives this
+ * domain a new broker and the replication server waits for an ack by CSN rather
+ * than by session, so an ack published after the hand-back reaches it all the
+ * same: the order of the two is not what this asserts.
+ */
+ final AckMsg ack = awaitAck(broker, csn);
+ assertTrue(ack.hasReplayError(),
+ "the ack of the abandoned delivery must report the replay error rather than"
+ + " be the plain ack a master would take for a durable write");
+ assertFalse(ack.hasTimeout(),
+ "the ack must be the one the abandoned delivery published, not the one the"
+ + " replication server makes up when it gives up waiting for it");
+ Assertions.assertThat(ack.getFailedServers())
+ .as("the replica which abandoned the delivery must be the one it names")
+ .containsExactly(domainSid);
+
+ /*
+ * Let the delivery which took over apply the change, and stop parking: an attempt
+ * of that delivery which came back on a lock it could not take would be parked
+ * again otherwise, with nothing left to release it.
+ */
+ parked.deregister();
+
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialApplied + 1,
+ "the change must be applied by the delivery which took over from the abandoned one");
+ /*
+ * A change applied twice goes through the expected count on its way up, so the
+ * value has to be seen to stay put rather than to be reached once - and for
+ * longer than the session restart which would bring that second delivery, or the
+ * assertion stops looking before what it is looking for could arrive.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-ok", initialApplied + 1,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "the change must be applied exactly once");
+ assertNull(DirectoryServer.getEntry(tmp.getName()), "the entry must have been deleted");
+
+ /*
+ * Abandoning a change is not giving up on it: the change is applied moments later,
+ * so nothing is counted as failed and the administrator is not told that this
+ * replica diverges.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was handed back must not be counted as one this replica gave up on");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE),
+ initialAlerts,
+ "a change which was handed back must not have this replica report a divergence");
+ }
+ finally
+ {
+ /*
+ * Whatever happened above: no replay thread of this server may be left parked,
+ * and the reconfiguration has to be over before the next one is started. The two
+ * would otherwise race for the pool of replay threads, which belongs to the
+ * server rather than to this test, and the one which creates its threads last
+ * drops the ones the other had just started without ever stopping them.
+ */
+ parked.deregister();
+ if (reconfiguration != null)
+ {
+ reconfiguration.join(SECONDS.toMillis(60));
+ reconfigurationFinished = !reconfiguration.isAlive();
+ }
+ if (reconfigurationFinished)
+ {
+ setNumberOfReplayThreads(null);
+ }
+ }
+ /*
+ * Read here rather than asserted while cleaning up, where a failure would replace
+ * the one the test was reporting. join() with a timeout returns the same whether
+ * the thread is over or not, and the pool of replay threads belongs to the server:
+ * a reconfiguration still inside stopReplayThreads() holds the monitor of
+ * MultimasterReplication, which restoring the pool would wait on for as long as an
+ * untimed join() takes, and one caught between stopping and creating orphans the
+ * threads the restore had just started. So the pool is left alone and this says so.
+ */
+ assertTrue(reconfigurationFinished,
+ "the replay thread reconfiguration never finished; the pool was left alone");
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Starts changing the number of replay threads of this server on a thread of its own.
+ * <p>
+ * It cannot run on the thread of the test: stopping the replay threads joins them, so it
+ * does not return while one of them is parked in a change, which is the whole point of
+ * this fixture.
+ *
+ * @param replayThreads how many replay threads the server must run, or {@code null} for
+ * as many as it computes on its own
+ * @param failure where the reconfiguration reports what it ran into, if anything
+ * @return the thread which is doing the reconfiguration
+ */
+ private static Thread startReplayThreadReconfiguration(
+ final Integer replayThreads, final AtomicReference<Throwable> failure)
+ {
+ final Thread reconfiguration = new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ setNumberOfReplayThreads(replayThreads);
+ }
+ catch (Throwable t)
+ {
+ failure.set(t);
+ }
+ }
+ }, "replay thread reconfiguration");
+ // A reconfiguration which never returns holds the monitor of MultimasterReplication:
+ // let the fork end on it rather than have it kept alive by a thread of this test.
+ reconfiguration.setDaemon(true);
+ reconfiguration.start();
+ return reconfiguration;
+ }
+
+ /**
+ * Reads the acknowledgement of the provided change off the broker it was published on.
+ * <p>
+ * The messages which come first are discarded: this broker is told about everything the
+ * replication server has for it, and the ack of one change is what is being looked for.
+ *
+ * @param broker the broker the change was published on
+ * @param csn the change the ack is expected for
+ * @return the ack of that change
+ * @throws Exception if it never arrived
+ */
+ private static AckMsg awaitAck(final ReplicationBroker broker, final CSN csn) throws Exception
+ {
+ final long deadline = System.nanoTime() + SECONDS.toNanos(60);
+ while (deadline - System.nanoTime() > 0)
+ {
+ final ReplicationMsg msg;
+ try
+ {
+ msg = broker.receive();
+ }
+ catch (SocketTimeoutException e)
+ {
+ // The broker reads under a timeout of its own, which is far shorter than the
+ // budget here: a quiet second is not an answer.
+ continue;
+ }
+ if (msg == null)
+ {
+ // The broker stopped rather than timed out: there is nothing left to read from,
+ // and reading it again would spin a core for the rest of the budget.
+ throw new AssertionError("the session " + csn + " was published on is gone,"
+ + " so the ack of that change can no longer arrive");
+ }
+ if (msg instanceof AckMsg && csn.equals(((AckMsg) msg).getCSN()))
+ {
+ return (AckMsg) msg;
+ }
+ }
+ throw new AssertionError("the delivery of " + csn + " was never acknowledged");
+ }
+
+ /**
+ * Waits for the delivery which took over from the abandoned one to be parked, reporting
+ * what the reconfiguration ran into when that is why nothing was parked.
+ * <p>
+ * A reconfiguration which throws once it has stopped the replay threads leaves this
+ * server with no replay thread at all: nothing can be parked then, and the timeout of
+ * the wait would be reported in place of the failure which brought it about.
+ *
+ * @param parked the park the delivery is expected to be caught in
+ * @param failure where the reconfiguration reports what it ran into, if anything
+ * @return the thread which is replaying the parked operation
+ * @throws Exception if no operation was parked in time
+ */
+ private static Thread awaitParkedOrReportReconfigurationFailure(
+ final ParkedReplay parked, final AtomicReference<Throwable> failure) throws Exception
+ {
+ try
+ {
+ return parked.awaitParked(60, SECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ final Throwable cause = failure.get();
+ if (cause == null)
+ {
+ throw e;
+ }
+ final AssertionError error = new AssertionError(
+ "the replay threads could not be reconfigured, so nothing was left to replay"
+ + " the change which was handed back", cause);
+ error.addSuppressed(e);
+ throw error;
+ }
+ }
+
+ /**
+ * Changes the number of replay threads of this server, the way a change of
+ * {@code num-update-replay-threads} does on a live server.
+ *
+ * @param replayThreads how many replay threads the server must run, or {@code null} for
+ * as many as it computes on its own
+ */
+ private static void setNumberOfReplayThreads(final Integer replayThreads)
+ {
+ final ReplicationSynchronizationProviderCfg cfg =
+ mock(ReplicationSynchronizationProviderCfg.class);
+ /*
+ * An unstubbed mock hands out 0 replay threads and no connection timeout, and both
+ * would outlive this test: the whole server shares one pool of replay threads. The
+ * number of them is what this test is changing; the connection timeout is read back
+ * from the running server, so that it is restored rather than restated.
+ */
+ when(cfg.getNumUpdateReplayThreads()).thenReturn(replayThreads);
+ when(cfg.getConnectionTimeout())
+ .thenReturn((long) MultimasterReplication.getConnectionTimeoutMS());
+ multimasterReplication().applyConfigurationChange(cfg);
+ }
+
+ /** Returns the replication synchronization provider of the running server. */
+ private static MultimasterReplication multimasterReplication()
+ {
+ // Read as an Object: the provider is declared with the configuration of its own type,
+ // which is not the one the registry lists.
+ for (Object provider : DirectoryServer.getSynchronizationProviders())
+ {
+ if (provider instanceof MultimasterReplication)
+ {
+ return (MultimasterReplication) provider;
+ }
+ }
+ throw new AssertionError("this server runs no replication synchronization provider");
+ }
+
+ /**
+ * Waits for the provided thread to be inside the {@code join()} of the replay threads it
+ * is stopping.
+ * <p>
+ * Waiting for that, rather than for the thread to be started, is what puts the change in
+ * the hands of a thread which has already been asked to stop: every replay thread is
+ * asked to stop before the first of them is joined, so a thread which is joining has
+ * asked the parked one. Where the thread waits is checked as well as that it waits: the
+ * state on its own would be satisfied by any wait at all, including one taken before the
+ * replay threads were asked to stop.
+ *
+ * @param reconfiguration the thread which is changing the number of replay threads
+ * @param failure where that thread reports what it ran into, if anything
+ * @throws Exception if it never reached the join
+ */
+ private static void awaitStoppingTheReplayThreads(
+ final Thread reconfiguration, final AtomicReference<Throwable> failure) throws Exception
+ {
+ final long deadline = System.nanoTime() + SECONDS.toNanos(60);
+ while (deadline - System.nanoTime() > 0)
+ {
+ if (reconfiguration.getState() == Thread.State.WAITING
+ && isJoiningTheReplayThreads(reconfiguration))
+ {
+ return;
+ }
+ if (!reconfiguration.isAlive())
+ {
+ if (failure.get() != null)
+ {
+ throw new AssertionError(
+ "the replay threads could not be reconfigured", failure.get());
+ }
+ fail("the replay threads were stopped without joining the one which holds a change");
+ }
+ // Sampling the stack of a running thread costs a handshake with it, so it is done
+ // often enough to open the gate promptly and not so often as to slow the server
+ // this test is watching: the thread it waits for is not going anywhere.
+ Thread.sleep(10);
+ }
+ fail("the reconfiguration never reached the join() of the replay threads");
+ }
+
+ /** Returns whether the provided thread is waiting on the replay threads it stopped. */
+ private static boolean isJoiningTheReplayThreads(final Thread reconfiguration)
+ {
+ for (final StackTraceElement frame : reconfiguration.getStackTrace())
+ {
+ if ("stopReplayThreads".equals(frame.getMethodName())
+ && MultimasterReplication.class.getName().equals(frame.getClassName()))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
* Test case for [Issue 889]: every change which can not be replayed must be given up
* on, not only the one which fails on its own.
* <p>
--
Gitblit v1.10.0