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/UpdateOperationTest.java | 764 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 763 insertions(+), 1 deletions(-)
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},
--
Gitblit v1.10.0