* 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 @@ private void recordChangeResolved(CSN csn) * 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. + *
+ * 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 @@ private void resetSessionRestartBackoff() * 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. + *
+ * 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 @@ private void skipUnreplayableChange(CSN csn, LocalizableMessage cause) { 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 @@ private void sendUnreplayedChangeAlert(LocalizableMessage cause) @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. + *
+ * 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). + *
+ * 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. + *
+ * 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. + *
+ * 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 @@ private boolean recoverFromReplayFailure( * 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 @@ public void requestSessionRestart() } /** - * 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. *
* 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 @@ public void disable() /* * 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 @@ public void enable() * 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 @@ protected void importBackend(InputStream input) throws DirectoryException * 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 c6b8fc1718..65f1ba9517 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 @@ ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port w 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 d57e0f99ca..123ace36a6 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 @@ public class UpdateOperationTest extends ReplicationTestCase */ 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"; @@ -3882,6 +3905,744 @@ public CSN getCSN() } } + /** + * Test case for [Issue 942]: a change whose replay keeps failing is warned about once + * per interval rather than once per delivery. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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
* 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 3dc590f719..0f0c15ccde 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.
+ *
+ * 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
@@ -325,6 +330,72 @@ public void aRequestWhichStoodWhileTheImportRanIsNotRunOnceItIsOver() throws Exc
}
}
+ /**
+ * A total update forgets the deliveries which were folded into no warning, along with
+ * the changes they were deliveries of (issue #942).
+ *
+ * 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.
+ *
+ * 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