From d07bb318976a8e1c65b1c26b431b0909380ce473 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 17 Sep 2026 11:57:18 +0000
Subject: [PATCH] [#942] Warn once per interval that a change is being retried, not once per delivery (#982)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 71 +++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 5
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 764 ++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 203 ++++++++++-
4 files changed, 1,019 insertions(+), 24 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 409083e..5e75be0 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
@@ -340,6 +340,14 @@
*/
private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000;
/**
+ * How long the warning telling that a change is being asked for again is not logged
+ * again, for the reason the alert above is not sent again and for one more: a change
+ * which keeps failing is asked for again every {@link #MAX_REPLAY_RETRY_DELAY_IN_MS}
+ * at the slowest, for as long as its give-up budget lasts - and how long that is has
+ * been the administrator's to set since issue #901.
+ */
+ private static final long REPLAY_RETRY_WARNING_INTERVAL_IN_MS = 60000;
+ /**
* What the ack of a delivery whose replay ran out of memory says did not apply the
* change.
* <p>
@@ -386,11 +394,14 @@
@GuardedBy("sessionRestartBackoff")
private long sessionRestartBackoffWakes;
/**
- * How many times in a row the session was restarted without a change being replayed in
- * between. The backoff is computed from this rather than from the failures of the
- * change which happens to open the recovery: an outage fails every change in flight,
- * and the ones which are sent for the first time would otherwise keep the wait at its
- * shortest for as long as the outage lasts.
+ * How many times in a row the session was restarted since this replica last replayed a
+ * change while nothing was failing, or was last disabled or imported into. A replay of
+ * another change while one still fails does not reset it, and neither does giving up on
+ * the last one failing: see {@link #resetReplayFailureTracking()} and
+ * {@link #skipUnreplayableChange(CSN, LocalizableMessage)}. The backoff is computed from
+ * this rather than from the failures of the change which happens to open the recovery:
+ * an outage fails every change in flight, and the ones which are sent for the first time
+ * would otherwise keep the wait at its shortest for as long as the outage lasts.
*/
private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger();
/**
@@ -481,14 +492,38 @@
*/
private volatile long replayDrainTimeoutInMs = REPLAY_DRAIN_TIMEOUT_IN_MS;
/**
- * Stands for "the alert about a change this replica gave up on was never sent". The
- * time it is compared with only moves forward from an origin which is arbitrary, so
- * zero is not far enough in the past to say it.
+ * How long this domain does not warn again about a change it asks for again. Only the
+ * tests, which can not wait out {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS} between two
+ * warnings, set another value.
*/
- private static final long UNREPLAYED_CHANGE_ALERT_NEVER_SENT = Long.MIN_VALUE / 2;
+ private volatile long replayRetryWarningIntervalInMs = REPLAY_RETRY_WARNING_INTERVAL_IN_MS;
+ /**
+ * Stands for "a replay failure was never reported yet", whether by the alert about a
+ * change this replica gave up on or by the warning about a change it asks for again.
+ * The time it is compared with only moves forward from an origin which is arbitrary,
+ * so zero is not far enough in the past to say it.
+ */
+ private static final long REPLAY_FAILURE_NEVER_REPORTED = Long.MIN_VALUE / 2;
/** When the alert about a change this replica gave up on was last sent. */
private final AtomicLong lastUnreplayedChangeAlertTime =
- new AtomicLong(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
+ new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED);
+ /** When the warning about a change this replica asks for again was last logged. */
+ private final AtomicLong lastReplayRetryWarningTime =
+ new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED);
+ /**
+ * How many failed deliveries were folded into no warning since the last one was logged,
+ * or since this replica last stopped failing: the next warning says how many it stands
+ * for. Forgotten when a change is replayed or given up on and nothing is failing anymore
+ * - see {@link #resetReplayFailureTracking()} and
+ * {@link #skipUnreplayableChange(CSN, LocalizableMessage)}, only the first of which
+ * forgets the session restart backoff with it - and with the pending changes when this
+ * domain is disabled or imported into, and once more when it is enabled back, for the
+ * delivery whose failure was being recorded while the domain went down: it is folded
+ * after {@link #disable()} forgot the count, see {@link #enable()}. A delivery which
+ * fails while the domain is shutting down, disabled or imported into is not counted: it
+ * is not warned about either, see {@link #sessionHasAnOwner()}.
+ */
+ private final AtomicInteger foldedReplayRetryWarnings = new AtomicInteger();
/**
* The result codes conflict resolution knows how to solve. The result code the server
* puts on an internal error is configurable, and every one of these reports a failure -
@@ -2371,7 +2406,7 @@
logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN);
return;
}
- resetSessionRestartBackoff();
+ resetReplayFailureTracking();
}
else
{
@@ -3598,12 +3633,13 @@
private void recordChangeResolved(CSN csn)
{
updateError(csn);
- resetSessionRestartBackoff();
+ resetReplayFailureTracking();
}
/**
* Has the change which fails next start the backoff between the session restarts over,
- * if this replica is not failing any change anymore.
+ * and the warning about it count the deliveries it stands for from zero, if this
+ * replica is not failing any change anymore.
* <p>
* A change made it and nothing is failing anymore, so the backend is serving again and
* the session is not being restarted in a row. While something is still failing, a
@@ -3611,12 +3647,19 @@
* applied here fails alone, among changes which replay perfectly well, and letting
* those reset the wait would have this domain tear its session down every second for as
* long as that one change takes to be given up on.
+ * <p>
+ * The deliveries which were folded into no warning go with the backoff rather than into
+ * the next warning: a line logged when this domain fails again - a day later, over
+ * another change - would have them read as deliveries of that failure. How long the
+ * warning is not logged again is deliberately left alone, so that a backend which fails
+ * and recovers in turn is not one warning per failure again.
*/
- private void resetSessionRestartBackoff()
+ private void resetReplayFailureTracking()
{
if (!remotePendingChanges.hasFailingChanges())
{
consecutiveSessionRestarts.set(0);
+ foldedReplayRetryWarnings.set(0);
}
}
@@ -3630,6 +3673,15 @@
* wait would have a replica which gives up on a change now and then ask for every
* change of an outage as fast as the replication server can send them, which is what
* {@link #consecutiveSessionRestarts} is there to prevent.
+ * <p>
+ * The deliveries folded into no warning are not: a warning logged over the next failure
+ * - a day later, on the default budget - would otherwise read as counting them. What is
+ * forgotten is a count, of deliveries which each had their trace line: the line which
+ * says the change is being skipped reports every delivery of this change when its
+ * budget was spent, and nothing reports the folded deliveries of the changes which were
+ * replayed meanwhile. They are forgotten when this was the last change failing, as they
+ * are when a change is replayed: while another change is still failing, they are
+ * deliveries of the outage the next warning is about.
*
* @param csn the CSN of the change which could not be replayed
* @param cause the message describing why it could not be replayed
@@ -3640,6 +3692,10 @@
{
numFailedReplayedUpdates.incrementAndGet();
sendUnreplayedChangeAlert(cause);
+ if (!remotePendingChanges.hasFailingChanges())
+ {
+ foldedReplayRetryWarnings.set(0);
+ }
}
// Otherwise the change is not listed as pending anymore - the domain was disabled
// while it was being replayed - so it has not been skipped: the replication server
@@ -3679,7 +3735,89 @@
@VisibleForTesting
public void resetUnreplayedChangeAlertThrottle()
{
- lastUnreplayedChangeAlertTime.set(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
+ lastUnreplayedChangeAlertTime.set(REPLAY_FAILURE_NEVER_REPORTED);
+ }
+
+ /**
+ * Warns that this replica could not replay a change and is asking for it again.
+ * <p>
+ * The warning is not logged again for {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS},
+ * for the reason the alert above is not sent again and for one more: a change which
+ * keeps failing is delivered again every {@link #MAX_REPLAY_RETRY_DELAY_IN_MS} at the
+ * slowest, so one line per delivery is the same warning every ten seconds, for as long
+ * as the give-up budget of the change lasts - a budget the administrator sets, and
+ * which can be unlimited (issue #942).
+ * <p>
+ * The deliveries which are not warned about are counted, so that the line which is
+ * logged next says how many of them it stands for, and they are traced for whoever
+ * turns the replication debug logging on.
+ *
+ * @param csn the CSN of the change which could not be replayed
+ * @param failure how long, and over how many deliveries, its replay has been failing
+ */
+ private void logReplayRetryWarning(CSN csn, RemotePendingChanges.ReplayFailure failure)
+ {
+ final long now = monotonicNowInMs();
+ final long lastLogged = lastReplayRetryWarningTime.get();
+ if (now - lastLogged >= replayRetryWarningIntervalInMs
+ && lastReplayRetryWarningTime.compareAndSet(lastLogged, now))
+ {
+ logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts(),
+ failure.getFailingForMs(), foldedReplayRetryWarnings.getAndSet(0));
+ }
+ else
+ {
+ /*
+ * A failure which loses the race against the thread which is logging right now is
+ * counted for the line being written when its increment lands before that thread
+ * takes the count, and for the next line otherwise. It is counted once either way:
+ * what the count says is how many deliveries went without a warning of their own,
+ * and which of two lines a minute apart says it says nothing else.
+ */
+ foldedReplayRetryWarnings.incrementAndGet();
+ logger.trace("Could not replay change %s in domain %s: delivery %d, failing for %d ms",
+ csn, getBaseDN(), failure.getAttempts(), failure.getFailingForMs());
+ }
+ }
+
+ /**
+ * Lets the next change whose replay fails be warned about straight away.
+ * <p>
+ * Only there for the tests which check the warning: they must not be at the mercy of
+ * the warning another test logged less than
+ * {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS} ago. The deliveries folded into no
+ * warning are left alone: when they are forgotten is this domain's to decide, and the
+ * tests check that it does.
+ */
+ @VisibleForTesting
+ public void resetReplayRetryWarningThrottle()
+ {
+ lastReplayRetryWarningTime.set(REPLAY_FAILURE_NEVER_REPORTED);
+ }
+
+ /**
+ * Returns how long this domain does not warn again about a change it asks for again.
+ *
+ * @return the interval in milliseconds
+ */
+ @VisibleForTesting
+ public long getReplayRetryWarningInterval()
+ {
+ return replayRetryWarningIntervalInMs;
+ }
+
+ /**
+ * Sets how long this domain does not warn again about a change it asks for again.
+ * <p>
+ * Only there for the tests which check the warning: they can not wait out
+ * {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS} between two of them.
+ *
+ * @param intervalInMs the interval in milliseconds
+ */
+ @VisibleForTesting
+ public void setReplayRetryWarningInterval(long intervalInMs)
+ {
+ replayRetryWarningIntervalInMs = intervalInMs;
}
/**
@@ -3801,9 +3939,12 @@
* 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.
+ * reason. The throttle is left alone as well - the trace line a folded delivery is
+ * written to asks for that memory too, and this delivery is not one which goes
+ * unlogged: the error ends the replay thread, and the uncaught exception handler of
+ * DirectoryThread writes the line and raises the alert for it.
*/
- logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
+ logReplayRetryWarning(csn, failure);
}
/*
* This change is not owned by anyone anymore, so the session has to be restarted for
@@ -4121,8 +4262,9 @@
}
/**
- * Returns how many times in a row the session was restarted without a change being
- * replayed in between: the count the backoff of the next restart is computed from.
+ * Returns how many times in a row the session was restarted while this replica kept
+ * failing - see {@link #consecutiveSessionRestarts} for what does and does not reset it:
+ * the count the backoff of the next restart is computed from.
* <p>
* Only there for the tests, which read it to see a restart reach its backoff rather than
* wait a delay out and hope it began: the count is bumped on the way into the wait, so
@@ -4917,10 +5059,15 @@
/*
* The recovery from a failed replay is over as well: the change it was asking for
* is gone with the pending changes, so a leftover request would have a replay thread
- * stop and start the session once for a delivery which can not come.
+ * stop and start the session once for a delivery which can not come. The deliveries
+ * folded into no warning go with it, or the first warning over the data loaded back
+ * would read as counting the deliveries of a change which is not listed anymore. A
+ * delivery whose failure is being recorded right now is folded after this, and is
+ * forgotten by enable() instead.
*/
sessionRestarts.clear();
consecutiveSessionRestarts.set(0);
+ foldedReplayRetryWarnings.set(0);
}
// Woken outside the lock it does not hold: a restart which is waiting has nothing
// left to wait for, the session it would start being one this domain is not serving.
@@ -5085,8 +5232,20 @@
* change which is gone with the pending changes. Every request standing here is that
* one: the domain has been disabled since anything could ask, and the session started
* below asks for everything the ServerState loaded below does not cover.
+ *
+ * The deliveries folded into no warning are forgotten here for the same window: a
+ * thread which is recording a failed replay reads the flag, then folds the delivery
+ * - recoverFromReplayFailure() logs before it asks - then asks, and runs what it
+ * asked for itself, which restartSession() turns down on a domain which owns its
+ * session. So on that road it is the fold rather than the request which outlives
+ * disable()'s clear, and the first warning over the data loaded back would count a
+ * delivery of a change which went with the pending changes. What reads the count is
+ * a warning, and none is logged between disable() and here short of that thread's
+ * own, so a test tells this zeroing and disable()'s apart by nothing: they stand or
+ * fall together.
*/
sessionRestarts.clear();
+ foldedReplayRetryWarnings.set(0);
try
{
loadDataState();
@@ -5597,11 +5756,13 @@
* listed a change meanwhile: the listener thread is the one running this import,
* and the replay threads gave up every attempt while the flag was set. The restart
* a replay thread may have asked for before the total update owned the session
- * goes with them: the caller starts the session again from the reloaded state.
+ * goes with them, as do the deliveries folded into no warning: the caller starts
+ * the session again from the reloaded state.
*/
remotePendingChanges.clear();
sessionRestarts.clear();
consecutiveSessionRestarts.set(0);
+ foldedReplayRetryWarnings.set(0);
importingData = false;
}
}
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 c6b8fc1..65f1ba9 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -617,8 +617,9 @@
NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \
in domain "%s" from this directory server DS(%d): rejecting the request from the remote directory server DS(%d): %s
WARN_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (delivery %d, each \
- attempted several times in place). The change has not been recorded as replayed: restarting the \
- session to the replication server so that it is sent again
+ attempted several times in place, failing for %d ms). The change has not been recorded as \
+ replayed: restarting the session to the replication server so that it is sent again. %d further \
+ deliveries failed in this domain without a warning of their own
ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s": its replay has been \
failing for %d ms over %d deliveries, each attempted several times in place. The change is being \
skipped: this replica now diverges from the rest of the topology and must be reinitialized. \
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 d57e0f9..123ace3 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
@@ -23,6 +23,7 @@
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.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
@@ -34,11 +35,15 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -114,6 +119,24 @@
*/
private static final String TEST_GIVE_UP_DELAY = "2000ms";
+ /**
+ * How long a change is retried in the test which checks the warning logged after this
+ * replica gave up on a change: long enough for a delivery to be folded into no warning
+ * before the budget is spent, with a session restart slower than usual allowed for -
+ * {@link #TEST_GIVE_UP_DELAY} is spent on the second delivery once the restart takes a
+ * second. In the duration syntax of the property.
+ */
+ private static final String TEST_GIVE_UP_DELAY_OVER_FOLDED_DELIVERIES = "4000ms";
+
+ /**
+ * How long the warning about a change being asked for again is not logged again, in the
+ * tests which check that warning. The session is left down for a second, then two, then
+ * three between the deliveries of a change which keeps failing: long enough for the
+ * first few of them to fall into one interval, short enough not to make a test wait out
+ * the minute of the server.
+ */
+ private static final long TEST_REPLAY_RETRY_WARNING_INTERVAL_IN_MS = 10000;
+
/** The configuration attribute which carries the replay give-up budget of a domain. */
private static final String ATTR_REPLAY_GIVE_UP_DELAY = "ds-cfg-replay-give-up-delay";
@@ -3883,6 +3906,744 @@
}
/**
+ * Test case for [Issue 942]: a change whose replay keeps failing is warned about once
+ * per interval rather than once per delivery.
+ * <p>
+ * The session is left down for ten seconds at the longest between two deliveries, so a
+ * warning per delivery is the same line every ten seconds for as long as the change is
+ * retried - and how long that is has been the administrator's to set since #901. The
+ * budget is unlimited here: this is the domain which must not go silent over a change it
+ * keeps asking for.
+ * <p>
+ * Each warning says how many deliveries were folded into it, and only those: the third
+ * warning stands for the deliveries since the second one, not for those since the first.
+ */
+ @Test
+ public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Exception
+ {
+ testSetUp("aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long interval = shortenReplayRetryWarningInterval(domain);
+ try
+ {
+ setReplayGiveUpDelay("unlimited");
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = addUserEntry("user.942.retried");
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
+
+ /*
+ * The session is left down for a second, then two, between the deliveries, so the
+ * first three of them fall into one interval: one warning, where there were three
+ * before this one was throttled.
+ */
+ waitForDeliveries(3);
+ assertEquals(replayRetryWarnings(csn).size(), 1,
+ "a change which keeps failing must be warned about once per interval, not once per delivery");
+
+ /*
+ * Once the interval has passed the change is warned about again, and the line
+ * which comes says how many deliveries were folded into no warning meanwhile.
+ */
+ waitForReplayRetryWarnings(csn, 2);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
+ .as("the warning must say how many failed deliveries it stands for")
+ .isGreaterThanOrEqualTo(1);
+ final int deliveriesAtSecondWarning = deliveriesSoFar();
+
+ /*
+ * The deliveries the second warning stands for are not the third one's as well: a
+ * count which was read rather than taken when the line was written would have every
+ * warning of an outage count every delivery since its first one. The deliveries
+ * since the second warning are the one which logged the third and those folded
+ * into it, so the count is below that number.
+ */
+ waitForReplayRetryWarnings(csn, 3);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(2)))
+ .as("the third warning must only stand for the deliveries since the second one")
+ .isLessThan(deliveriesSoFar() - deliveriesAtSecondWarning);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, csn, failed);
+ throw failed;
+ }
+ letTheChangeBeCovered(domain, csn);
+ }
+ finally
+ {
+ // What can not throw first: a cleanup which throws skips the ones after it.
+ domain.setReplayRetryWarningInterval(interval);
+ broker.stop();
+ resetReplayGiveUpDelay();
+ }
+ }
+
+ /**
+ * Test case for [Issue 942]: the deliveries folded into no warning are forgotten when
+ * this replica stops failing, rather than carried over to the next failure.
+ * <p>
+ * The count is what the next warning says it stands for. A warning logged over another
+ * change, once the backend has served again for a while, must not read as counting the
+ * deliveries of the failure before it.
+ */
+ @Test
+ public void aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore() throws Exception
+ {
+ testSetUp("aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore");
+ logger.error(LocalizableMessage.raw("Starting replication test : "
+ + "aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long interval = shortenReplayRetryWarningInterval(domain);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry first = addUserEntry("user.942.recovered");
+ Entry second = addUserEntry("user.942.failing.next");
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ // Two deliveries which fail: the first is warned about, the second is folded.
+ broker.publish(new DeleteMsg(first.getName(), csn, getEntryUUID(first.getName())));
+ waitForDeliveries(2);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, csn, failed);
+ throw failed;
+ }
+ // The backend serves again: the delivery which comes next replays the change.
+ letTheChangeBeCovered(domain, csn);
+
+ /*
+ * Another change fails, and the interval is not waited out: only the timestamp of the
+ * throttle is put back, the count is the domain's to keep or to forget.
+ */
+ domain.resetReplayRetryWarningThrottle();
+ final CSN later = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(second.getName(), later, getEntryUUID(second.getName())));
+ waitForReplayRetryWarnings(later, 1);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(later).get(0)))
+ .as("the warning over a new failure must not count the deliveries of the one before")
+ .isEqualTo(0);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, later, failed);
+ throw failed;
+ }
+ letTheChangeBeCovered(domain, later);
+ }
+ finally
+ {
+ domain.setReplayRetryWarningInterval(interval);
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 942]: giving up on the change which was the last one failing
+ * forgets the deliveries folded into no warning as well.
+ * <p>
+ * The line which says the change is being skipped reports every delivery it had, so
+ * nothing is lost, and the next failure - a day later, on the default budget - must not
+ * be warned about as if the deliveries of the change given up on were its own.
+ */
+ @Test
+ public void aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries() throws Exception
+ {
+ testSetUp("aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries");
+ logger.error(LocalizableMessage.raw("Starting replication test : "
+ + "aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long interval = shortenReplayRetryWarningInterval(domain);
+ try
+ {
+ setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_OVER_FOLDED_DELIVERIES);
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = addUserEntry("user.942.given.up");
+ final CSN csn = gen.newCSN();
+ final CSN later = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
+ /*
+ * The change is given up on once its budget is spent, which the ServerState
+ * covering a change which was never applied says. The budget outlasts the first
+ * deliveries, so at least one of them was folded into no warning by then: the
+ * delivery which spends the budget is neither warned about nor folded.
+ */
+ waitUntilCovered(domain, csn);
+ Assertions.assertThat(deliveriesSoFar())
+ .as("the budget must have outlasted a delivery which was folded into no warning")
+ .isGreaterThanOrEqualTo(3);
+
+ /*
+ * Another change fails, and the interval is not waited out: only the timestamp of
+ * the throttle is put back, the count is the domain's to keep or to forget.
+ */
+ domain.resetReplayRetryWarningThrottle();
+ broker.publish(new DeleteMsg(tmp.getName(), later, getEntryUUID(tmp.getName())));
+ waitForReplayRetryWarnings(later, 1);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(later).get(0)))
+ .as("the warning after a change was given up on must not count its deliveries")
+ .isEqualTo(0);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, later, failed);
+ throw failed;
+ }
+ /*
+ * Replayed now that the backend serves again, or given up on: either way it is
+ * covered. The budget is still the short one here, and nothing spends it: a change is
+ * only given up on over a delivery which failed, and none fails once the short circuit
+ * is gone.
+ */
+ letTheChangeBeCovered(domain, later);
+ }
+ finally
+ {
+ // What can not throw first: a cleanup which throws skips the ones after it.
+ domain.setReplayRetryWarningInterval(interval);
+ broker.stop();
+ resetReplayGiveUpDelay();
+ }
+ }
+
+ /**
+ * Test case for [Issue 942]: disabling the domain - for an LDIF import, a restore, or a
+ * backend being taken offline - forgets the deliveries folded into no warning, along
+ * with the changes they were deliveries of.
+ * <p>
+ * The changes listed as pending do not outlive the ServerState the domain saves on its
+ * way down, and the recovery from a failed replay goes with them: the first warning
+ * over the data loaded back must not count the deliveries of a change which is not
+ * listed anymore.
+ */
+ @Test
+ public void aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore() throws Exception
+ {
+ testSetUp("aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore");
+ logger.error(LocalizableMessage.raw("Starting replication test : "
+ + "aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long interval = shortenReplayRetryWarningInterval(domain);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = addUserEntry("user.942.disabled");
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
+ /*
+ * Three deliveries which fail: the first is warned about, the second is folded
+ * for sure by the time the third is delivered - a delivery is folded once its
+ * attempts are over, and the next one only comes over the session restarted
+ * after that.
+ */
+ waitForDeliveries(3);
+
+ /*
+ * The domain goes down and comes back, the way an import or a restore has it: the
+ * change is not listed anymore, and the replication server sends it again over the
+ * new session, from the ServerState which was saved without it. The throttle is put
+ * back while the domain is down, when nothing fails, so that the first failure over
+ * the data loaded back is warned about straight away - with the count the domain
+ * kept or forgot.
+ */
+ domain.disable();
+ domain.resetReplayRetryWarningThrottle();
+ domain.enable();
+ waitForReplayRetryWarnings(csn, 2);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
+ .as("the first warning after the domain was disabled must not count the deliveries before")
+ .isEqualTo(0);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, csn, failed);
+ throw failed;
+ }
+ letTheChangeBeCovered(domain, csn);
+ }
+ finally
+ {
+ domain.setReplayRetryWarningInterval(interval);
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 942]: a change replayed while another one keeps failing does not
+ * forget the deliveries folded into no warning.
+ * <p>
+ * This is what the issue looks like live: one entry which can not be applied here, among
+ * changes which replay perfectly well. Each of those is a change replayed, and the count
+ * is forgotten when a change is replayed - only when nothing is failing anymore, though.
+ * The deliveries folded so far are deliveries of the outage the next warning is about, and
+ * a warning which said {@code 0 further} over them would be a wrong number which looks
+ * right.
+ */
+ @Test
+ public void aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing()
+ throws Exception
+ {
+ testSetUp("aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing");
+ logger.error(LocalizableMessage.raw("Starting replication test : "
+ + "aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ /*
+ * The interval is left as the server has it, a minute, and only the throttle is put back:
+ * the second warning must be the one let through below, once the other change has been
+ * replayed, so that its count says what the domain kept over that replay. A warning the
+ * interval let through meanwhile would take the count with it.
+ */
+ domain.resetReplayRetryWarningThrottle();
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = addUserEntry("user.942.failing.alone");
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
+ /*
+ * Three deliveries which fail: the first is warned about, the second and the third
+ * are folded. Two folded at the least, so that a count which was kept is told apart
+ * from the one delivery which may be folded between the other change being replayed
+ * and the throttle being put back.
+ *
+ * Waited for by the first attempt of the fourth delivery rather than by the last
+ * attempt of the third: a delivery is folded once its attempts in place are spent,
+ * after the last of them tripped the short circuit, and the fourth delivery only
+ * comes over the session restart the third asks for once it has been folded. A
+ * count read on that last attempt would credit a fold which is not there yet, and
+ * the throttle put back below before it lands would have the third delivery warned
+ * about rather than folded - with the one fold there was.
+ */
+ waitForDeliveryAttempts(3 * IN_PLACE_REPLAY_ATTEMPTS + 1);
+ // Deliveries 2 and 3: delivery 1 was warned about.
+ final int foldedBeforeTheReplay = 2;
+
+ /*
+ * Another entry is added while the delete keeps failing. An add does not depend on
+ * the delete of another entry, so it is replayed - a change made while the delete
+ * is still listed as failing.
+ */
+ final String otherUUID = "94200000-0000-0000-0000-000000000001";
+ final Entry other = TestCaseUtils.makeEntry(
+ "dn: uid=user.942.replayed.meanwhile," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.942.replayed.meanwhile",
+ "cn: Aaccf Amar",
+ "sn: Amar",
+ "entryUUID: " + otherUUID);
+ broker.publish(addMsg(gen, other, otherUUID, baseUUID));
+ assertNotNull(getEntry(other.getName(), 10000, true),
+ "the change of another entry must be replayed while the delete keeps failing");
+ /*
+ * The backoff of the session restarts is kept under the same guard as the count:
+ * three deliveries failed, so three restarts ran in a row - the fourth delivery came
+ * over the third - and the replay of the add forgot none of them.
+ */
+ Assertions.assertThat(domain.getConsecutiveSessionRestarts())
+ .as("a replay of another change must not forget the backoff of the one still failing")
+ .isGreaterThanOrEqualTo(3);
+
+ /*
+ * The throttle is put back, so that the next delivery of the delete is warned about
+ * with the count the domain kept - or forgot - over the replay.
+ */
+ domain.resetReplayRetryWarningThrottle();
+ waitForReplayRetryWarnings(csn, 2);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
+ .as("a replay of another change must not forget the deliveries of the one still failing")
+ .isGreaterThanOrEqualTo(foldedBeforeTheReplay);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, csn, failed);
+ throw failed;
+ }
+ letTheChangeBeCovered(domain, csn);
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 942]: giving up on a change while another one keeps failing does
+ * not forget the deliveries folded into no warning either.
+ * <p>
+ * The road a change is given up on when its budget is spent is the road a change no
+ * operation can be built from takes at its first delivery, and that one needs no budget
+ * to be waited out: a message whose modifications can not be decoded is given up on while
+ * the delete which keeps failing is still listed as failing.
+ */
+ @Test
+ public void givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing()
+ throws Exception
+ {
+ testSetUp("givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing");
+ logger.error(LocalizableMessage.raw("Starting replication test : "
+ + "givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing"));
+
+ final int serverId = 19;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ // The interval is left as the server has it, for the reason the case above gives.
+ domain.resetReplayRetryWarningThrottle();
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+ Entry tmp = addUserEntry("user.942.failing.alone.too");
+ Entry other = addUserEntry("user.942.given.up.meanwhile");
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+ try
+ {
+ broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName())));
+ // Three deliveries which fail, two of them folded, waited for as in the case above.
+ waitForDeliveryAttempts(3 * IN_PLACE_REPLAY_ATTEMPTS + 1);
+ final int foldedBeforeTheGiveUp = 2;
+
+ /*
+ * A change of another entry which no operation can be built from is given up on at
+ * its first delivery, while the delete is still listed as failing. The count of the
+ * changes this replica gave up on says when it has been.
+ */
+ final long givenUpBefore = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ broker.publish(
+ undecodableModifyMsg(gen.newCSN(), other.getName(), getEntryUUID(other.getName())));
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", givenUpBefore + 1,
+ "the change which can not be decoded must be given up on while the delete keeps failing");
+
+ domain.resetReplayRetryWarningThrottle();
+ waitForReplayRetryWarnings(csn, 2);
+ Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1)))
+ .as("giving up on another change must not forget the deliveries of the one still failing")
+ .isGreaterThanOrEqualTo(foldedBeforeTheGiveUp);
+ }
+ catch (Throwable failed)
+ {
+ letTheChangeBeCovered(domain, csn, failed);
+ throw failed;
+ }
+ letTheChangeBeCovered(domain, csn);
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Shortens how long the domain does not warn again about a change it asks for again, and
+ * returns the interval to put back in a finally: the domain outlives the test methods,
+ * and none of them can afford the minute of the server.
+ * <p>
+ * The throttle is the domain's too: a change another test was retrying less than an
+ * interval ago would have the first warning of this one folded into its own, so it is put
+ * back as well.
+ *
+ * @param domain the domain of the test
+ * @return the interval the domain had, in milliseconds
+ */
+ private static long shortenReplayRetryWarningInterval(LDAPReplicationDomain domain)
+ {
+ final long interval = domain.getReplayRetryWarningInterval();
+ domain.setReplayRetryWarningInterval(TEST_REPLAY_RETRY_WARNING_INTERVAL_IN_MS);
+ domain.resetReplayRetryWarningThrottle();
+ return interval;
+ }
+
+ /** Adds an entry with the provided uid below the base DN, the entry the change fails on. */
+ private Entry addUserEntry(String uid) throws Exception
+ {
+ return TestCaseUtils.addEntry(
+ "dn: uid=" + uid + "," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: " + uid,
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ }
+
+ /**
+ * Returns how many deliveries of the delete which can not be replayed have failed since
+ * the short circuit was registered: a delivery is attempted
+ * {@link LDAPReplicationDomain#IN_PLACE_REPLAY_ATTEMPTS} times in place before the change
+ * is asked for again, and each attempt trips the short circuit once.
+ *
+ * @return the number of deliveries whose attempts in place are all spent
+ */
+ private static int deliveriesSoFar()
+ {
+ return ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")
+ / IN_PLACE_REPLAY_ATTEMPTS;
+ }
+
+ /**
+ * Waits until the delete which can not be replayed has been delivered, and failed, the
+ * provided number of times.
+ *
+ * @param deliveries how many deliveries to wait for
+ * @throws Exception if the deliveries do not come
+ */
+ private static void waitForDeliveries(final int deliveries) throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(deliveriesSoFar() >= deliveries,
+ "the change was not delivered again after its replay failed: " + deliveriesSoFar()
+ + " deliveries where " + deliveries + " were expected");
+ }
+ });
+ }
+
+ /**
+ * Waits until the delete which can not be replayed has been attempted the provided number
+ * of times, over however many deliveries.
+ * <p>
+ * The first attempt of a delivery is one more than the attempts of the deliveries before
+ * it, and it is what tells that the delivery before it has been warned about or folded:
+ * a delivery is folded once its attempts in place are spent, after the last of them
+ * tripped the short circuit, and the next delivery only comes over the session restart
+ * asked for once it has been. {@link #waitForDeliveries(int)} returns on that last
+ * attempt, before the fold.
+ *
+ * @param attempts how many attempts to wait for
+ * @throws Exception if the attempts do not come
+ */
+ private static void waitForDeliveryAttempts(final int attempts) throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ final int attemptsSoFar =
+ ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse");
+ assertTrue(attemptsSoFar >= attempts,
+ "the change was not attempted again after its replay failed: " + attemptsSoFar
+ + " attempts where " + attempts + " were expected");
+ }
+ });
+ }
+
+ /**
+ * Waits until this replica has warned the provided number of times that the provided
+ * change is being asked for again.
+ *
+ * @param csn the CSN of the change whose replay keeps failing
+ * @param warnings how many warnings to wait for
+ * @throws Exception if the warnings do not come
+ */
+ private static void waitForReplayRetryWarnings(final CSN csn, final int warnings)
+ throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertEquals(replayRetryWarnings(csn).size(), warnings,
+ "a change which keeps failing was not warned about " + warnings + " times");
+ }
+ });
+ }
+
+ /**
+ * Takes the short circuit back and waits until the change it was failing is covered by
+ * the ServerState - replayed now that the backend serves again, or given up on - so that
+ * a case does not leave its change to the next one: a change left failing here would be
+ * the next test's, holding its ServerState back, its session restart backoff up and the
+ * warnings of that test folded into its own.
+ * <p>
+ * Called at the end of a case, and from its catch with the failure when it has one,
+ * rather than from a finally: a wait which expired in a finally would replace the
+ * assertion it was cleaning up after.
+ *
+ * @param domain the domain of the test
+ * @param csn the CSN of the change
+ * @throws Exception if the change is not covered
+ */
+ private static void letTheChangeBeCovered(final LDAPReplicationDomain domain, final CSN csn)
+ throws Exception
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ waitUntilCovered(domain, csn);
+ }
+
+ /**
+ * {@link #letTheChangeBeCovered(LDAPReplicationDomain, CSN)} for a case which failed: the
+ * failure is what is thrown, and what went wrong here is added to it.
+ *
+ * @param domain the domain of the test
+ * @param csn the CSN of the change
+ * @param failed the failure of the case
+ */
+ private static void letTheChangeBeCovered(
+ final LDAPReplicationDomain domain, final CSN csn, final Throwable failed)
+ {
+ try
+ {
+ letTheChangeBeCovered(domain, csn);
+ }
+ catch (Throwable late)
+ {
+ failed.addSuppressed(late);
+ }
+ }
+
+ /**
+ * Waits until the ServerState of the domain covers the provided change: it was replayed
+ * once the backend served again, or given up on.
+ *
+ * @param domain the domain of the test
+ * @param csn the CSN of the change
+ * @throws Exception if the change is not covered
+ */
+ private static void waitUntilCovered(final LDAPReplicationDomain domain, final CSN csn)
+ throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "the change must be replayed once the backend serves again, or given up on");
+ }
+ });
+ }
+
+ /**
+ * Returns how many deliveries the provided warning says were folded into no warning of
+ * their own.
+ *
+ * @param warning a warning about a change being asked for again
+ * @return the count the warning carries
+ */
+ private static int foldedDeliveriesSaidBy(String warning)
+ {
+ final Matcher count = Pattern.compile("(\\d+) further deliveries").matcher(warning);
+ assertTrue(count.find(),
+ "the warning does not say how many deliveries it stands for: " + warning);
+ return Integer.parseInt(count.group(1));
+ }
+
+ /**
+ * Returns the warnings this replica logged about the provided change being asked for
+ * again, oldest first.
+ * <p>
+ * The error log of the test server is written to a writer which keeps every record, so
+ * the warnings about one change are the records which carry the ordinal of the message
+ * and the CSN of the change.
+ * <p>
+ * The test server registers two error log publishers over that one writer, so it keeps
+ * every record twice, each copy timestamped by its publisher: a warning is two records
+ * which read the same once the timestamp is left out - the two publishers read the clock
+ * one after the other, and a second which turns over between the two reads would have
+ * one warning counted as two by the records as they are. So what is returned here is the
+ * messages which differ, each as many times as half its records: two warnings which read
+ * the same are two warnings, which happens when the domain was disabled in between and
+ * asks for the change again as one it does not remember.
+ *
+ * @param csn the CSN of the change whose replay keeps failing
+ * @return the warnings which name it, in the order they were first logged
+ */
+ private static List<String> replayRetryWarnings(CSN csn)
+ {
+ final String messageId = "msgID=" + WARN_REPLAY_RETRYING_CHANGE.ordinal();
+ final Map<String, Integer> records = new LinkedHashMap<>();
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ if (record.contains(messageId) && record.contains(csn.toString()))
+ {
+ records.merge(record.substring(record.indexOf(" msg=")), 1, Integer::sum);
+ }
+ }
+ final List<String> warnings = new ArrayList<>();
+ for (Map.Entry<String, Integer> message : records.entrySet())
+ {
+ for (int i = 0; i < (message.getValue() + 1) / 2; i++)
+ {
+ warnings.add(message.getKey());
+ }
+ }
+ return warnings;
+ }
+
+ /**
* 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.
@@ -4351,7 +5112,8 @@
* <p>
* The domain outlives the test methods, so a test which shortens the budget puts it back
* with {@link #resetReplayGiveUpDelay()} in a finally, and calls this one before that
- * try: the reset then only ever runs on an attribute which is there to be removed.
+ * try or first inside it - the reset then runs on an attribute which is there to be
+ * removed, or finds it gone already, which it takes for the default it was asking for.
*
* @param delay
* the budget in the duration syntax of the property: {@code 2000ms},
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
index 3dc590f..0f0c15c 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -37,6 +37,7 @@
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
import org.opends.server.replication.protocol.InitializeRequestMsg;
@@ -70,6 +71,10 @@
* The exporter is a broker of this test, so that the test says when the entries arrive: the
* change is replayed while the import is waiting for them - or, for the request, while the
* exporter is holding the answer.
+ * <p>
+ * The {@code timeOut} each case declares is what it is expected to take at the most; it is
+ * not what bounds it. {@code TestListener} sets the timeout of every test method from the
+ * {@code org.opends.test.timeout} property, ten minutes under Maven and none outside it.
*/
@SuppressWarnings("javadoc")
public class ReplayDuringImportTest extends ReplicationTestCase
@@ -326,6 +331,72 @@
}
/**
+ * A total update forgets the deliveries which were folded into no warning, along with
+ * the changes they were deliveries of (issue #942).
+ * <p>
+ * The changes listed as pending do not outlive the ServerState the import replaces, and
+ * the recovery from a failed replay goes with them - the session restart backoff, and the
+ * count the next warning about a change being asked for again says it stands for. The
+ * first warning over the imported data must not count the deliveries of a change which
+ * is not listed anymore.
+ * <p>
+ * Nothing sends a change of this test again - the exporter never had it - so the count
+ * is fed by two changes failing within one interval rather than by one change delivered
+ * twice: the first is warned about, the second is folded into no warning. The changes
+ * are deletes: a short circuit on the modifies would be tripped by the ServerState being
+ * saved to the base entry and by the import disabling the backend it replaces.
+ */
+ @Test(timeOut = 120_000)
+ public void aWarningAfterTheImportDoesNotCountTheDeliveriesBefore() throws Exception
+ {
+ final Entry warnedAbout = TestCaseUtils.addEntry(
+ "dn: cn=warnedAbout," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: warnedAbout",
+ "sn: warnedAbout");
+ final Entry folded = TestCaseUtils.addEntry(
+ "dn: cn=folded," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: folded",
+ "sn: folded");
+ final String warnedAboutUUID = getEntryUUID(warnedAbout.getName());
+ final String foldedUUID = getEntryUUID(folded.getName());
+ final String[] exported = exportedEntries();
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new DeleteMsg(warnedAbout.getName(), gen.newCSN(), warnedAboutUUID));
+ replayMsg(new DeleteMsg(folded.getName(), gen.newCSN(), foldedUUID));
+
+ startImportInto(exported.length);
+ finishImport(exported);
+
+ /*
+ * Only the timestamp of the throttle is put back, so that the failure over the
+ * imported data is warned about straight away: the count is the domain's to keep or
+ * to forget.
+ */
+ domain.resetReplayRetryWarningThrottle();
+ final CSN csn = gen.newCSN();
+ replayMsg(new DeleteMsg(DN.valueOf(IMPORTED_ENTRY_DN), csn, IMPORTED_ENTRY_UUID));
+ final List<String> warnings = errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn);
+ assertThat(warnings).as("the change which fails over the imported data must be warned about")
+ .isNotEmpty();
+ assertThat(warnings.get(0))
+ .as("the first warning after the import must not count the deliveries before it")
+ .contains(" 0 further deliveries");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+
+ /**
* Has the exporter start a total update into this replica, and returns once the backend
* of the domain is deregistered for it: from then on the import is reading the session,
* and a change replayed here is replayed into no backend.
--
Gitblit v1.10.0