From d8eda1f037ab8d2be25b0415c630623bced003a4 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 09:27:24 +0300 Subject: [PATCH 1/5] [#942] Warn once per interval that a change is being retried, not once per delivery Fixes #942. recoverFromReplayFailure() logged WARN_REPLAY_RETRYING_CHANGE for every delivery of a change whose replay failed. The session is left down for ten seconds at the longest between two deliveries, so a change which keeps failing had the same line logged every ten seconds for as long as it was retried - and since #901 how long that is belongs to the administrator, "unlimited" included. The throttle takes the shape of the alert next to it, a timestamp and a CAS: one line per domain and per minute. Per domain rather than per change, unlike what the issue sketched - the cause which makes one change unreplayable makes every change in flight unreplayable, and a replica whose ServerState is held back by the barrier change is sent every change which follows it over and over, so a per-change throttle would still leave one line per change accumulated since the outage began. The deliveries which are not logged are counted rather than dropped: the line which is logged says how many of them it stands for and how long the change has been failing, and the folded ones are traced for whoever turns replication debug logging on. The count goes back to zero where the session restart backoff does, so that a line logged over another failure a day later does not read as counting its deliveries; how long the warning is not logged again is deliberately left alone, or a backend which fails and recovers in turn is one warning per failure again. resetSessionRestartBackoff() is resetReplayFailureTracking() now that it clears both, and UNREPLAYED_CHANGE_ALERT_NEVER_SENT is REPLAY_FAILURE_NEVER_REPORTED, the one origin the alert and the warning are both measured from. --- .../plugin/LDAPReplicationDomain.java | 108 ++++++++++-- .../opends/messages/replication.properties | 5 +- .../replication/UpdateOperationTest.java | 157 ++++++++++++++++++ 3 files changed, 256 insertions(+), 14 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 409083ed20..a724f74f55 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 @@ -339,6 +339,14 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * in flight unreplayable, and one alert per change would be a storm. */ 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. @@ -481,14 +489,24 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { */ 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. + * 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 UNREPLAYED_CHANGE_ALERT_NEVER_SENT = Long.MIN_VALUE / 2; + 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 not warned about since the last warning was logged. + * They are counted rather than dropped: the line which is logged next says how many + * deliveries it stands for. + */ + 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 +2389,7 @@ void synchronize(PostOperationOperation op) logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN); return; } - resetSessionRestartBackoff(); + resetReplayFailureTracking(); } else { @@ -3598,12 +3616,13 @@ private static long monotonicNowInMs() 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. *

* 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 +3630,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); } } @@ -3679,7 +3705,62 @@ 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 rather than dropped, so the + * line which is logged 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 >= REPLAY_RETRY_WARNING_INTERVAL_IN_MS + && 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 next line rather than for the one being written: what the count + * says is how many deliveries went unlogged, and carrying one of them over to the + * next interval 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. + */ + @VisibleForTesting + public void resetReplayRetryWarningThrottle() + { + lastReplayRetryWarningTime.set(REPLAY_FAILURE_NEVER_REPORTED); + foldedReplayRetryWarnings.set(0); } /** @@ -3801,9 +3882,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 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..319a25dbea 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 being logged since the previous warning 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..89b9882bbb 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,6 +35,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Set; import java.util.concurrent.TimeoutException; @@ -3882,6 +3884,161 @@ 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. + */ + @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); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry tmp = TestCaseUtils.addEntry( + "dn: uid=user.942," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.942", + "cn: Aaccf Amar", + "sn: Amar"); + String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + /* + * The throttle is the domain's and the domain outlives the test methods: a change + * another test was retrying less than an interval ago would have the first warning + * of this one folded into its own. + */ + domain.resetReplayRetryWarningThrottle(); + final CSN csn = gen.newCSN(); + try + { + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + + /* + * A delivery burns IN_PLACE_REPLAY_ATTEMPTS short circuits before the session is + * restarted and the change is asked for again, so three times that many of them + * are three deliveries which failed - and three warnings, before this one was + * throttled. The give-up budget is minutes and the interval is a minute, so the + * change is still being retried by then and the deliveries all fall into one + * interval. + */ + TestTimer timer = new TestTimer.Builder() + .maxSleep(60, SECONDS) + .sleepTimes(100, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse") + > 3 * IN_PLACE_REPLAY_ATTEMPTS, + "the change was not delivered again after its replay failed"); + } + }); + 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: a domain which + * never gives up - the budget can be unlimited - must not go silent over a change + * it is still asking for, and the line which comes says how many deliveries went + * unlogged in the meantime. + */ + TestTimer intervalTimer = new TestTimer.Builder() + .maxSleep(120, SECONDS) + .sleepTimes(500, MILLISECONDS) + .toTimer(); + intervalTimer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertEquals(replayRetryWarnings(csn).size(), 2, + "a change which is still failing an interval later must be warned about again"); + } + }); + Assertions.assertThat(replayRetryWarnings(csn).get(1)) + .as("the warning must say how many failed deliveries it stands for") + .containsPattern("[1-9]\\d* further deliveries failed"); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + } + + /* + * The backend serves again, so the delivery which comes next replays the change: a + * change left failing here would be the next test's, holding its ServerState back + * and its session restart backoff up. + */ + TestTimer replayTimer = new TestTimer.Builder() + .maxSleep(60, SECONDS) + .sleepTimes(200, MILLISECONDS) + .toTimer(); + replayTimer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + assertTrue(domain.getServerState().cover(csn), + "the change must be replayed once the backend serves again"); + } + }); + } + finally + { + broker.stop(); + } + } + + /** + * 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: what is returned here is the records which differ. Two warnings + * about the same change never read the same - the delivery they report, how long the + * change has been failing and how many deliveries were folded into them all move on. + * + * @param csn the CSN of the change whose replay keeps failing + * @return the warnings which name it, in the order they were logged + */ + private static List replayRetryWarnings(CSN csn) + { + final String messageId = "msgID=" + WARN_REPLAY_RETRYING_CHANGE.ordinal(); + final Set warnings = new LinkedHashSet<>(); + for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) + { + if (record.contains(messageId) && record.contains(csn.toString())) + { + warnings.add(record); + } + } + return new ArrayList<>(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 From 5a620425eed8c28dceb19b0f6dcb8ee0d88aef6b Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 15 Sep 2026 13:42:23 +0300 Subject: [PATCH 2/5] [#942] Forget the folded deliveries wherever the backoff is forgotten, and pin every reset Review round 2. The count of deliveries folded into no warning went back to zero where a change was replayed and nothing was failing anymore, and nowhere else: disable() cleared the pending changes and the session restart backoff and left the count, as did the end of a total update - the same triple, landed by #968 after the review - and so did giving up on the change which was the last one failing. The first warning over the next failure, a day later, then read as counting the deliveries of the failure before. The count goes with the backoff in disable() and after the import now, and on the give-up road it goes alone, under the guard the recovery reset has: the backoff deliberately stays there, and ERR_REPLAY_SKIPPING_CHANGE reports every delivery the change had, so nothing is lost. Neither zeroing of the count was pinned - the test reset zeroed it itself. It resets the timestamp only now, the interval is a @VisibleForTesting knob the way the drain timeout is, and five cases pin one mechanism each, watched red on the mutant which removes it: the third warning of a change which keeps failing stands for the deliveries since the second one only, and the first warning over a new failure says 0 further after a change was replayed, after one was given up on, after the domain was disabled and enabled, and after a total update. The warning says "without a warning of their own" rather than "without being logged" - the folded deliveries are traced - and the test reads it by message rather than by record: the two error log publishers of the test server read the clock one after the other, so a second turning over between them would count one warning as two. Each message is counted half as many times as it is kept, since two warnings read the same to the letter once the domain has forgotten the change and asks for it again. --- .../plugin/LDAPReplicationDomain.java | 75 ++- .../opends/messages/replication.properties | 2 +- .../replication/UpdateOperationTest.java | 513 +++++++++++++++--- .../plugin/ReplayDuringImportTest.java | 67 +++ 4 files changed, 560 insertions(+), 97 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 a724f74f55..e0a8ebf9bb 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 @@ -488,6 +488,12 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * another value. */ private volatile long replayDrainTimeoutInMs = REPLAY_DRAIN_TIMEOUT_IN_MS; + /** + * 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 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. @@ -502,9 +508,13 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { private final AtomicLong lastReplayRetryWarningTime = new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED); /** - * How many failed deliveries were not warned about since the last warning was logged. - * They are counted rather than dropped: the line which is logged next says how many - * deliveries it stands for. + * How many failed deliveries were not warned about since the last warning was logged, + * or since this replica last stopped failing. They are counted rather than dropped: the + * line which is logged next says how many deliveries it stands for. They are forgotten + * along with the session restart backoff - when a change is replayed or given up on and + * nothing is failing anymore, see {@link #resetReplayFailureTracking()} and + * {@link #skipUnreplayableChange(CSN, LocalizableMessage)} - and along with the pending + * changes, when this domain is disabled or imported into. */ private final AtomicInteger foldedReplayRetryWarnings = new AtomicInteger(); /** @@ -3656,6 +3666,13 @@ private void resetReplayFailureTracking() * 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: the line which says the change is being + * skipped reports every delivery it had, so nothing is lost by forgetting them, and a + * warning logged over the next failure - a day later, on the default budget - would + * otherwise read as counting them. 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 @@ -3666,6 +3683,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 @@ -3729,7 +3750,7 @@ private void logReplayRetryWarning(CSN csn, RemotePendingChanges.ReplayFailure f { final long now = monotonicNowInMs(); final long lastLogged = lastReplayRetryWarningTime.get(); - if (now - lastLogged >= REPLAY_RETRY_WARNING_INTERVAL_IN_MS + if (now - lastLogged >= replayRetryWarningIntervalInMs && lastReplayRetryWarningTime.compareAndSet(lastLogged, now)) { logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts(), @@ -3739,9 +3760,10 @@ private void logReplayRetryWarning(CSN csn, RemotePendingChanges.ReplayFailure f { /* * A failure which loses the race against the thread which is logging right now is - * counted for the next line rather than for the one being written: what the count - * says is how many deliveries went unlogged, and carrying one of them over to the - * next interval says nothing else. + * 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", @@ -3754,13 +3776,39 @@ private void logReplayRetryWarning(CSN csn, RemotePendingChanges.ReplayFailure f *

* 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. + * {@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); - foldedReplayRetryWarnings.set(0); + } + + /** + * 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; } /** @@ -5001,10 +5049,13 @@ 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. */ 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. @@ -5681,11 +5732,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 319a25dbea..65f1ba9517 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -619,7 +619,7 @@ NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \ WARN_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (delivery %d, each \ 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 being logged since the previous warning + 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 89b9882bbb..bea7ff0080 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 @@ -35,12 +35,15 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; -import java.util.LinkedHashSet; +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; @@ -116,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"; @@ -3890,7 +3911,12 @@ public CSN getCSN() *

* 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. + * 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 @@ -3902,113 +3928,418 @@ public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Except final int serverId = 19; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + final long interval = shortenReplayRetryWarningInterval(domain); + setReplayGiveUpDelay("unlimited"); try { CSNGenerator gen = new CSNGenerator(serverId, 0); - - Entry tmp = TestCaseUtils.addEntry( - "dn: uid=user.942," + baseDN, - "objectClass: top", - "objectClass: person", - "objectClass: organizationalPerson", - "objectClass: inetOrgPerson", - "uid: user.942", - "cn: Aaccf Amar", - "sn: Amar"); - String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); - - final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); - /* - * The throttle is the domain's and the domain outlives the test methods: a change - * another test was retrying less than an interval ago would have the first warning - * of this one folded into its own. - */ - domain.resetReplayRetryWarningThrottle(); + Entry tmp = addUserEntry("user.942.retried"); final CSN csn = gen.newCSN(); + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); try { - ShortCircuitPlugin.registerShortCircuit( - OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); - broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName()))); /* - * A delivery burns IN_PLACE_REPLAY_ATTEMPTS short circuits before the session is - * restarted and the change is asked for again, so three times that many of them - * are three deliveries which failed - and three warnings, before this one was - * throttled. The give-up budget is minutes and the interval is a minute, so the - * change is still being retried by then and the deliveries all fall into one - * interval. + * 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. */ - TestTimer timer = new TestTimer.Builder() - .maxSleep(60, SECONDS) - .sleepTimes(100, MILLISECONDS) - .toTimer(); - timer.repeatUntilSuccess(new CallableVoid() - { - @Override - public void call() throws Exception - { - assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse") - > 3 * IN_PLACE_REPLAY_ATTEMPTS, - "the change was not delivered again after its replay failed"); - } - }); + 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: a domain which - * never gives up - the budget can be unlimited - must not go silent over a change - * it is still asking for, and the line which comes says how many deliveries went - * unlogged in the meantime. + * 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. */ - TestTimer intervalTimer = new TestTimer.Builder() - .maxSleep(120, SECONDS) - .sleepTimes(500, MILLISECONDS) - .toTimer(); - intervalTimer.repeatUntilSuccess(new CallableVoid() - { - @Override - public void call() throws Exception - { - assertEquals(replayRetryWarnings(csn).size(), 2, - "a change which is still failing an interval later must be warned about again"); - } - }); - Assertions.assertThat(replayRetryWarnings(csn).get(1)) + waitForReplayRetryWarnings(csn, 2); + Assertions.assertThat(foldedDeliveriesSaidBy(replayRetryWarnings(csn).get(1))) .as("the warning must say how many failed deliveries it stands for") - .containsPattern("[1-9]\\d* further deliveries failed"); + .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); } finally { ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + waitUntilCovered(domain, csn); + } + } + finally + { + resetReplayGiveUpDelay(); + domain.setReplayRetryWarningInterval(interval); + broker.stop(); + } + } + + /** + * 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); + } + finally + { + // The backend serves again: the delivery which comes next replays the change. + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + waitUntilCovered(domain, csn); } /* - * The backend serves again, so the delivery which comes next replays the change: a - * change left failing here would be the next test's, holding its ServerState back - * and its session restart backoff up. + * 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. */ - TestTimer replayTimer = new TestTimer.Builder() - .maxSleep(60, SECONDS) - .sleepTimes(200, MILLISECONDS) - .toTimer(); - replayTimer.repeatUntilSuccess(new CallableVoid() + domain.resetReplayRetryWarningThrottle(); + final CSN later = gen.newCSN(); + ShortCircuitPlugin.registerShortCircuit( + OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue()); + try { - @Override - public void call() throws Exception - { - assertTrue(domain.getServerState().cover(csn), - "the change must be replayed once the backend serves again"); - } - }); + 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); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + waitUntilCovered(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); + setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_OVER_FOLDED_DELIVERIES); + try + { + 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); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + resetReplayGiveUpDelay(); + // Replayed now that the backend serves again, or given up on: either way it is covered. + waitUntilCovered(domain, later); + } + } + finally + { + domain.setReplayRetryWarningInterval(interval); + broker.stop(); + } + } + + /** + * 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); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + waitUntilCovered(domain, csn); + } + } + finally + { + domain.setReplayRetryWarningInterval(interval); + 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 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"); + } + }); + } + + /** + * Waits until the ServerState of the domain covers the provided change: it was replayed + * once the backend served again, or given up on. + *

+ * Called from the finally which takes the short circuit back, so that a case which fails + * 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. + * + * @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. @@ -4018,25 +4349,37 @@ public void call() throws Exception * and the CSN of the change. *

* The test server registers two error log publishers over that one writer, so it keeps - * every record twice: what is returned here is the records which differ. Two warnings - * about the same change never read the same - the delivery they report, how long the - * change has been failing and how many deliveries were folded into them all move on. + * 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 logged + * @return the warnings which name it, in the order they were first logged */ private static List replayRetryWarnings(CSN csn) { final String messageId = "msgID=" + WARN_REPLAY_RETRYING_CHANGE.ordinal(); - final Set warnings = new LinkedHashSet<>(); + final Map records = new LinkedHashMap<>(); for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) { if (record.contains(messageId) && record.contains(csn.toString())) { - warnings.add(record); + records.merge(record.substring(record.indexOf(" msg=")), 1, Integer::sum); + } + } + final List warnings = new ArrayList<>(); + for (Map.Entry message : records.entrySet()) + { + for (int i = 0; i < (message.getValue() + 1) / 2; i++) + { + warnings.add(message.getKey()); } } - return new ArrayList<>(warnings); + return warnings; } /** 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..87d2d043eb 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; @@ -325,6 +326,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 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, From 239e317f64541740a99a9b9760205c02df6031b2 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 16 Sep 2026 11:25:23 +0300 Subject: [PATCH 3/5] [#942] Pin the guard which keeps the count while another change is still failing Review round 3. The count of deliveries folded into no warning is zeroed when a change is replayed or given up on, and only when nothing is failing anymore: one entry this replica can not apply fails alone, among changes which replay perfectly well, and each of those is a change replayed. No case reached the guard's true arm - at every reset the failing change was the only one - so a count zeroed on every commit stayed green. Two cases now, one per zeroing, each red on the mutant which drops its guard. A delete keeps failing over three deliveries, two of them folded; an add of another entry is replayed meanwhile, or a modify whose modifications can not be decoded is given up on at its first delivery - the same road a spent budget takes, with no budget to wait out - and the warning let through after it says at least the two folds. Two rather than one, since the window between the other change and the reset of the throttle can take one fold, and did: the unguarded count read 1 there. The interval is left at the server's minute in these two, since a warning it let through in between would take the count with it. The give-up budget is set first inside the try whose finally resets it, after the interval is put back and the broker stopped, and the change of a case is released at its end and from its catch - a wait which expires in a finally would replace the assertion it was cleaning up after. The javadoc of the count says what every reset does and what a delivery which fails while the session has an owner is not, the import test says what its timeOut is, and "nothing is lost" says what is: a count, of deliveries which each had their trace. --- .../plugin/LDAPReplicationDomain.java | 36 +-- .../replication/UpdateOperationTest.java | 261 ++++++++++++++++-- .../plugin/ReplayDuringImportTest.java | 4 + 3 files changed, 258 insertions(+), 43 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 e0a8ebf9bb..e5b20a2128 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 @@ -508,13 +508,15 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { private final AtomicLong lastReplayRetryWarningTime = new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED); /** - * How many failed deliveries were not warned about since the last warning was logged, - * or since this replica last stopped failing. They are counted rather than dropped: the - * line which is logged next says how many deliveries it stands for. They are forgotten - * along with the session restart backoff - when a change is replayed or given up on and - * nothing is failing anymore, see {@link #resetReplayFailureTracking()} and - * {@link #skipUnreplayableChange(CSN, LocalizableMessage)} - and along with the pending - * changes, when this domain is disabled or imported into. + * 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. 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(); /** @@ -3667,12 +3669,14 @@ private void resetReplayFailureTracking() * 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: the line which says the change is being - * skipped reports every delivery it had, so nothing is lost by forgetting them, and a - * warning logged over the next failure - a day later, on the default budget - would - * otherwise read as counting them. 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. + * 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 @@ -3739,9 +3743,9 @@ public void resetUnreplayedChangeAlertThrottle() * 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 rather than dropped, so the - * line which is logged says how many of them it stands for, and they are traced for - * whoever turns the replication debug logging on. + * 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 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 bea7ff0080..65e0034905 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 @@ -3930,9 +3930,9 @@ public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Except openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); final long interval = shortenReplayRetryWarningInterval(domain); - setReplayGiveUpDelay("unlimited"); try { + setReplayGiveUpDelay("unlimited"); CSNGenerator gen = new CSNGenerator(serverId, 0); Entry tmp = addUserEntry("user.942.retried"); final CSN csn = gen.newCSN(); @@ -3973,17 +3973,19 @@ public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Except .as("the third warning must only stand for the deliveries since the second one") .isLessThan(deliveriesSoFar() - deliveriesAtSecondWarning); } - finally + catch (Throwable failed) { - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - waitUntilCovered(domain, csn); + letTheChangeBeCovered(domain, csn, failed); + throw failed; } + letTheChangeBeCovered(domain, csn); } finally { - resetReplayGiveUpDelay(); + // What can not throw first: a cleanup which throws skips the ones after it. domain.setReplayRetryWarningInterval(interval); broker.stop(); + resetReplayGiveUpDelay(); } } @@ -4021,12 +4023,13 @@ public void aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore() thr broker.publish(new DeleteMsg(first.getName(), csn, getEntryUUID(first.getName()))); waitForDeliveries(2); } - finally + catch (Throwable failed) { - // The backend serves again: the delivery which comes next replays the change. - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - waitUntilCovered(domain, csn); + 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 @@ -4044,11 +4047,12 @@ public void aWarningOverANewFailureDoesNotCountTheDeliveriesOfTheOneBefore() thr .as("the warning over a new failure must not count the deliveries of the one before") .isEqualTo(0); } - finally + catch (Throwable failed) { - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - waitUntilCovered(domain, later); + letTheChangeBeCovered(domain, later, failed); + throw failed; } + letTheChangeBeCovered(domain, later); } finally { @@ -4077,9 +4081,9 @@ public void aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries() throws E openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); final long interval = shortenReplayRetryWarningInterval(domain); - setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_OVER_FOLDED_DELIVERIES); 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(); @@ -4111,18 +4115,20 @@ public void aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries() throws E .as("the warning after a change was given up on must not count its deliveries") .isEqualTo(0); } - finally + catch (Throwable failed) { - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - resetReplayGiveUpDelay(); - // Replayed now that the backend serves again, or given up on: either way it is covered. - waitUntilCovered(domain, later); + letTheChangeBeCovered(domain, later, failed); + throw failed; } + // Replayed now that the backend serves again, or given up on: either way it is covered. + letTheChangeBeCovered(domain, later); } finally { + // What can not throw first: a cleanup which throws skips the ones after it. domain.setReplayRetryWarningInterval(interval); broker.stop(); + resetReplayGiveUpDelay(); } } @@ -4182,11 +4188,12 @@ public void aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore() t .as("the first warning after the domain was disabled must not count the deliveries before") .isEqualTo(0); } - finally + catch (Throwable failed) { - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - waitUntilCovered(domain, csn); + letTheChangeBeCovered(domain, csn, failed); + throw failed; } + letTheChangeBeCovered(domain, csn); } finally { @@ -4195,6 +4202,167 @@ public void aWarningAfterTheDomainWasDisabledDoesNotCountTheDeliveriesBefore() t } } + /** + * 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, and a delivery is folded before the session is restarted for the next + * one. 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. + */ + waitForDeliveries(3); + final int foldedBeforeTheReplay = deliveriesSoFar() - 1; + + /* + * 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 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, as in the case above. + waitForDeliveries(3); + final int foldedBeforeTheGiveUp = deliveriesSoFar() - 1; + + /* + * 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, @@ -4294,14 +4462,52 @@ public void call() throws Exception }); } + /** + * 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. - *

- * Called from the finally which takes the short circuit back, so that a case which fails - * 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. * * @param domain the domain of the test * @param csn the CSN of the change @@ -4851,7 +5057,8 @@ private static int indexOf(byte[] haystack, byte[] needle) *

* 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 87d2d043eb..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 @@ -71,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 From 07dc4d4358913f993de91f3f512fec14c8b8221f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 16 Sep 2026 14:39:45 +0300 Subject: [PATCH 4/5] [#942] Read the folded deliveries once the third one has been folded, and pin the backoff guard Review round 4. The two cases which pin the !hasFailingChanges() guards read how many deliveries were folded on the last in-place attempt of the third delivery, before it was folded: a delivery is folded once its attempts are spent, after the last of them tripped the short circuit. A replay thread which lagged past the throttle being put back had the third delivery warned about rather than folded, and the second warning said 1 where the case read 2 - a rare red on a loaded runner, never a false green. Both cases now wait for the first attempt of the fourth delivery, which only comes over the session restart the third asks for once it has been folded, and the premise is the constant it is: deliveries 2 and 3. The reviewer's probe, 500 ms at the top of logReplayRetryWarning(), is red 2/2 on the head's cases and green 2/2 on these; both guards dropped stays red 2/2. The backoff of the session restarts is kept under the same guard as the count, and nothing pinned that: the replay case reads getConsecutiveSessionRestarts() - #981's, in the rebase - after the add is replayed, three restarts having run in a row, and is red on the backoff zeroed on every commit. The give-up case says why its change is waited for under the short budget - nothing spends it, a change is only given up on over a delivery which failed - rather than putting the budget back first. --- .../replication/UpdateOperationTest.java | 75 ++++++++++++++++--- 1 file changed, 65 insertions(+), 10 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 65e0034905..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 @@ -4120,7 +4120,12 @@ public void aWarningAfterAChangeWasGivenUpOnDoesNotCountItsDeliveries() throws E letTheChangeBeCovered(domain, later, failed); throw failed; } - // Replayed now that the backend serves again, or given up on: either way it is covered. + /* + * 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 @@ -4244,13 +4249,21 @@ public void aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing 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, and a delivery is folded before the session is restarted for the next - * one. 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. + * 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. */ - waitForDeliveries(3); - final int foldedBeforeTheReplay = deliveriesSoFar() - 1; + 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 @@ -4271,6 +4284,14 @@ public void aReplayOfAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailing 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 @@ -4329,9 +4350,9 @@ public void givingUpOnAnotherChangeDoesNotForgetTheDeliveriesOfTheOneStillFailin try { broker.publish(new DeleteMsg(tmp.getName(), csn, getEntryUUID(tmp.getName()))); - // Three deliveries which fail, two of them folded, as in the case above. - waitForDeliveries(3); - final int foldedBeforeTheGiveUp = deliveriesSoFar() - 1; + // 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 @@ -4436,6 +4457,40 @@ public void call() throws Exception }); } + /** + * 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. From 4ee1f79c8921472e46a134e736d9fcc4733c1f1c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 16 Sep 2026 18:05:41 +0300 Subject: [PATCH 5/5] [#942] Forget a fold which landed after disable() where enable() forgets the request, and say what resets the backoff Review round 5. A thread which is recording a failed replay reads whether the session has an owner, then folds the delivery into the next warning, then asks for the session restart - and runs what it asked for itself, which restartSession() turns down on a domain which owns its session. One preempted across disable() folds after the count was zeroed with the pending changes, so the first warning over the data loaded back would count a delivery of a change which is not listed anymore. enable() forgets the count next to the request #981 clears there for the same window. Nothing reads the count in between, so the disable/enable case pins the two zeroings as a pair: both deleted is red, either alone is green, and the window itself is a preemption no test can hit on purpose. The javadocs of consecutiveSessionRestarts and of getConsecutiveSessionRestarts() said the count is reset by any change replayed in between. It has been kept while another change still fails since #892 put the !hasFailingChanges() guard on the reset, the commit the field's javadoc came in with; the accessor's, from #981, copied the wording. Both say now what does and does not reset it. --- .../plugin/LDAPReplicationDomain.java | 42 ++++++++++++++----- 1 file changed, 31 insertions(+), 11 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 e5b20a2128..5e75be0a39 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 @@ -394,11 +394,14 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { @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(); /** @@ -514,9 +517,11 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * - 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. 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()}. + * 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(); /** @@ -4257,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 @@ -5055,7 +5061,9 @@ public void disable() * 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. 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. + * 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); @@ -5224,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();