From 06ace4f681ce650ca12994f924b2efc392181e6f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 09:57:28 +0300 Subject: [PATCH] [#954] Give back the changes a replay which is unwound parked as dependencies A change which waits for another one is parked and stays owned by the replay thread which parked it: getNextUpdate() is what hands it out again, to whichever thread clears the change it was waiting for. A replay which is unwound leaves that thread without the road back - it takes the next delivery off the shared queue - so the change stayed owned by a thread which never came back to it, while every redelivery of a change a replay thread owns is refused as a duplicate. On a domain which then went quiet that change was where this replica's ServerState, and every change behind it from every master, stopped. RemotePendingChanges gives back the changes the calling thread parked and unparks them in the same step, under both locks, so that only one road can hand a change out: a change released while it is still listed as waiting would be handed to the thread getNextUpdate() gives it to and to the thread which takes over the delivery which follows. The changes another thread parked are left alone, as everywhere else. replay() gives them back before the road of the change it was replaying runs, since that road restarts the session and a change which is still owned when the replication server sends it again is turned down. They are handed back without a failure counted against them - they were never applied here - and the session is restarted for them. Which change this thread owns is read before they are given back, and not after. The read is a plain map lookup which allocates nothing, and the give-back below it allocates - it builds the list of what it released and the line which reports each one. A throw from it on the road it exists for, a JVM which has just refused an allocation, would otherwise reach the last resort of replay() with nothing read, and the change this thread was replaying would be left listed, uncommitted and owned by a thread which is ending: the wedge #922 is about, one road over. The order of the roads is unchanged - the parked changes are still handed back before the road of the change this thread was replaying restarts the session. Rebased onto the head of #958, itself rebased onto master. That branch grew a fifth commit, "[#922] Pin the ownership index on the hand-out and the give-back, and assert the alert an OutOfMemoryError leaves behind": the javadoc of the ownership index conflicted where both branches rewrote it and is merged, and the comment that commit wrote on the OutOfMemoryError arm of the ack - the give-back finds nothing once the change was committed - now says which parked changes the give-back does hand back on that road. Master's #928 steps over a modify whose entry DN does not parse rather than asking for it again, so the end-to-end test builds the change it parks behind from an operation which is built and then refused, the way the tests of #958 do since that merge; it takes the configured replay give-up budget, since #901 replaced the setters it used. --- .../plugin/LDAPReplicationDomain.java | 127 +++++++--- .../plugin/RemotePendingChanges.java | 89 ++++++- .../opends/messages/replication.properties | 3 + .../replication/UpdateOperationTest.java | 223 +++++++++++++++++- .../plugin/RemotePendingChangesTest.java | 150 +++++++++++- 5 files changed, 544 insertions(+), 48 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 b94a4dc8ba..9a496c0b58 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 @@ -2606,20 +2606,25 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * failing and is eventually given up on: handing it back bare would have this domain * ask for it, and restart its session for it, for as long as the server is up. * - * The changes this thread parked as waiting for another change are left alone: they - * are handed to whichever thread clears the change they are waiting for, and that - * thread takes them over. + * The changes this thread parked as waiting for another change are given back too, + * and before the road above runs: that road restarts the session, and a change which + * is still owned when the replication server sends it again over it is turned down as + * a duplicate - the one delivery which could have taken it over (issue #954). * - * Which change this thread owns is read before anything is done with it, and that - * read takes no lock and allocates nothing: everything below is gated on the answer, - * so a lookup which threw in its turn - on the road out of a JVM which has just - * refused an allocation - would leave the change listed, uncommitted and owned by a - * thread which is about to end, which is the state this whole issue is about. + * Which change this thread owns is read first of all, and that read takes no lock and + * allocates nothing: everything below is gated on the answer, so a lookup which threw + * in its turn - on the road out of a JVM which has just refused an allocation - would + * leave the change listed, uncommitted and owned by a thread which is about to end, + * which is the state this whole issue is about. It is read before the parked changes + * are given back rather than after, because that give-back allocates and can throw on + * the same road, and the last resort below can only hand back a change it was told + * about. */ CSN owned = null; try { owned = remotePendingChanges.getChangeOwnedByCurrentThread(); + final boolean parkedGivenBack = giveBackParkedChanges(); if (owned != null) { if (replayThreadShutdown.get() || shutdown.get() || disabled) @@ -2650,6 +2655,17 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError); } } + if (parkedGivenBack && !shutdown.get() && !disabled) + { + /* + * The road the change this thread was replaying took may have run the restart the + * give-back asked for - they ask for the same one - and it may have had none to + * run: this thread owned no change, or the change it owned was given up on. Run + * what is still requested, or a domain which then goes quiet would leave the + * changes which were handed back waiting for a delivery nobody asks for. + */ + runRequestedSessionRestarts(!replayThreadShutdown.get()); + } } catch (Throwable recoveryFailure) { @@ -2682,19 +2698,25 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) { suppress(recoveryFailure, reportFailure); } - try - { - runRequestedSessionRestarts(false); - } - catch (Throwable restartFailure) - { - /* - * Nothing is left to try: the change is listed, uncommitted and unowned, so any - * later session restart of this domain delivers it again. This goes with the - * throwable which is rethrown below rather than being reported on its own. - */ - suppress(recoveryFailure, restartFailure); - } + } + try + { + /* + * Outside the guard above, since the give-back of the changes this thread parked + * asks for the same restart and may be what threw: a change which nobody owns + * anymore is one only a new delivery brings back, whichever of the two roads + * released it (issue #954). + */ + runRequestedSessionRestarts(false); + } + catch (Throwable restartFailure) + { + /* + * Nothing is left to try: the changes are listed, uncommitted and unowned, so any + * later session restart of this domain delivers them again. This goes with the + * throwable which is rethrown below rather than being reported on its own. + */ + suppress(recoveryFailure, restartFailure); } // The error which unwound the replay is the one reported, whatever the give-back // ran into on top of it. @@ -3173,12 +3195,14 @@ else if (op instanceof ModifyDNOperation) * an OutOfMemoryError of its own, still owns its change: it is given back * counted, and the thread ends on this error rather than on the one it stepped * over. A replay which committed owns nothing anymore - commit() cleared the - * owner, and the index the give-back reads, in the same step - so the give-back - * is a no-op, and rightly so: a change which is in the data is not one to ask - * for again. What that road steps over is getNextUpdate() below, so the changes - * parked behind the committed change wait for the next replay of this domain to - * hand them out. That is the trade #923 asks for: a thread which met this error - * is not to carry on, not even for them. + * owner, and the index the give-back reads, in the same step - so the change it + * was replaying is not given back, and rightly so: a change which is in the data + * is not one to ask for again. What that road steps over is getNextUpdate() + * below, which hands out the changes parked behind the committed change: the + * ones this thread parked are given back on the way out of replay() and the + * session is restarted for them (issue #954), the ones other threads parked wait + * for the next replay of this domain to hand them out. That is the trade #923 + * asks for: a thread which met this error is not to carry on, not even for them. */ throw e; } @@ -3672,6 +3696,55 @@ private void runRequestedSessionRestarts(boolean wait) } } + /** + * Gives back the changes this replay thread parked as waiting for another change, on the + * way out of a replay which was unwound. + *

+ * A parked change is handed out again by {@code getNextUpdate()} alone, which every + * replay loop of this domain runs once it is done with a change: a parked change is + * replayed by whichever thread clears the change it was waiting for. A thread whose + * replay was unwound is not on that road anymore - it takes the next delivery off the + * replay queue - so a change it parked would be left owned by a thread which is not + * coming back to it, while every redelivery of it is refused as a duplicate. On a domain + * which then goes quiet that change is where this replica's ServerState, and every change + * behind it from every master, stops (issue #954). + *

+ * They are handed back without a failure being counted against them: they were never + * applied here, so the give-up budget which decides when this replica skips a change it + * can not apply is not this delivery's to spend, the way it is not for a change abandoned + * by a replay thread which is stopping. + *

+ * The delivery which carried one published no ack - the ack of a parked change is + * published by the delivery which replays it - so it is counted as processed here, the + * way a delivery which is dropped rather than replayed is: that count is of the + * deliveries this replica took off the session, and these are over. The window they hold + * is not given back either, and does not need to be: the session they came over is about + * to be restarted, and a session which starts is given its receive window anew. + * + * @return whether any change was handed back, so that the caller restarts the session for + * them: a change which nobody owns is one only a new delivery brings back + */ + private boolean giveBackParkedChanges() + { + final List parked = remotePendingChanges.releaseParkedChangesOwnedByCurrentThread(); + if (parked.isEmpty()) + { + return false; + } + /* + * Asked for before the changes are reported: a throw out of the report - the JVM which + * unwound this replay is out of memory - must not lose the restart which is what brings + * them back. + */ + sessionRestartRequested.set(true); + for (CSN csn : parked) + { + incProcessedUpdates(); + logger.info(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK, csn, getBaseDN()); + } + return true; + } + /** * Gives a change back to the replication server when this replay thread stops before it * could apply it. diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java index ec3ef7b756..b16ada2f3e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java @@ -17,7 +17,11 @@ */ package org.opends.server.replication.plugin; +import static java.util.Collections.*; + +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.NoSuchElementException; import java.util.SortedMap; import java.util.SortedSet; @@ -90,8 +94,10 @@ final class RemotePendingChanges * this issue is about (issue #922). *

* A thread is entered here when it takes a change over and removed when it gives it back, - * applies it, or parks it as waiting for another change - the parked ones are handed to - * whichever thread clears what they wait for, so they are not this one's to give back. + * applies it, or parks it as waiting for another change - a parked change is not the one + * this thread is replaying, and giving it back is + * {@link #releaseParkedChangesOwnedByCurrentThread()}, which reads the changes which are + * waiting rather than this index (issue #954). *

* The entry of a thread is written by that thread and by nobody else, and that - not the * lock - is what keeps the writes apart: the park in {@link #addDependency(PendingChange)} @@ -545,9 +551,10 @@ public boolean markInProgress(LDAPUpdateMsg msg) * Returns the CSN of the change the calling thread is replaying, when it still owns one. *

* A thread owns the change it is replaying and the ones it parked as waiting for another - * change. The parked ones are left out: they are handed to whichever thread clears the - * change they are waiting for, and that thread takes them over, so giving one back here - * would have the same change handed to two threads (issue #922). + * change. The parked ones are left out: they are not the change this thread is replaying, + * and giving one back is more than dropping its owner - it has to be unparked in the same + * step, or it would be handed out by two roads at once, which is what + * {@link #releaseParkedChangesOwnedByCurrentThread()} does (issues #922 and #954). *

* It is a plain read of {@link #changeBeingReplayed}: no lock is taken and nothing is * allocated. This is what the give-back on the way out of an unwound replay asks first, @@ -569,6 +576,72 @@ CSN getChangeOwnedByCurrentThread() return changeBeingReplayed.get(Thread.currentThread()); } + /** + * Gives back the changes the calling thread parked as waiting for another change, and + * takes them out of the changes which are waiting in the same step. + *

+ * A parked change stays owned by the thread which parked it while that thread goes on + * to the changes which follow: {@link #getNextUpdate()} is what hands it out again, to + * whichever replay thread clears the change it was waiting for, and that thread takes it + * over. A replay which is unwound leaves the thread which parked it without that road - + * it takes the next delivery off the replay queue instead - so the change would be left + * owned by a thread which is never coming back to it, and every redelivery of a change a + * replay thread owns is refused as a duplicate (issue #954). + *

+ * Unparking a change and giving it back is one step, under both locks, so that only one + * road can hand it out: a change which was released while it is still listed as waiting + * would be handed to the thread {@link #getNextUpdate()} gives it to and to the thread + * which takes over the delivery which follows - the double replay the ownership is there + * to prevent (OPENDJ-1115). + *

+ * The changes stay listed and uncommitted, and stay among the changes the newer ones are + * checked against, the way a change whose replay failed does: they are not in the data, + * so they hold this domain's ServerState back and the changes which follow them keep + * waiting for them. + *

+ * The changes another thread parked are left alone: a change is given back by the thread + * which owns it and by nobody else (issue #922). That thread may be inside the dependency + * checks which parked it - they park a change once per dependency it has - so a change + * released under it would be listed as waiting again a moment later, and handed out while + * the delivery which took it over is being replayed. + * + * @return the CSNs of the changes it gave back, oldest first; empty when this thread has + * no parked change left, which is what every replay which was not unwound while + * it held one leaves behind + */ + List releaseParkedChangesOwnedByCurrentThread() + { + final Thread current = Thread.currentThread(); + pendingChangesWriteLock.lock(); + dependentChangesLock.lock(); + try + { + if (dependentChanges.isEmpty()) + { + // Nothing is waiting, which is the state every replay but a handful leaves behind. + return emptyList(); + } + final List released = new ArrayList<>(); + final Iterator it = dependentChanges.iterator(); + while (it.hasNext()) + { + final PendingChange change = it.next(); + if (change.isOwnedBy(current)) + { + it.remove(); + change.setOwner(null); + released.add(change.getCSN()); + } + } + return released; + } + finally + { + dependentChangesLock.unlock(); + pendingChangesWriteLock.unlock(); + } + } + /** * Get the first update in the list that have some dependencies cleared. *

@@ -672,8 +745,10 @@ private void addDependency(PendingChange dependentChange) * parked one is handed to the thread which clears what it waits for, and one which is * not listed here anymore is gone with the pending changes of a domain which was * disabled. The owner stays as it is - it is what has getNextUpdate() hand the change - * over rather than leave it to nobody - and the give-back on the way out of an - * unwound replay leaves it alone (issue #922). + * over rather than leave it to nobody - and the give-back of the change a replay was + * unwound on leaves it alone (issue #922). What hands a parked change back is + * releaseParkedChangesOwnedByCurrentThread(), which unparks it in the same step so + * that the two roads can not hand it out at once (issue #954). */ changeBeingReplayed.remove(Thread.currentThread(), dependentChange.getCSN()); } 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 d5a7e69970..efd51f64c5 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -668,6 +668,9 @@ ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \ server after the replay which owned it was unwound: %s. The change has been released without its \ failure being counted, and the session is being restarted so that the change is delivered again +NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK_318=Change %s in domain "%s" was waiting for another change \ + to be replayed when the replay which parked it was unwound. The change has not been recorded as \ + replayed and is given back to the replication server, which still owns it and sends it again WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \ the replay of one of its changes to finish. A change which reaches the backend from now on \ is not recorded in the ServerState being saved, so the replication server sends it again \ 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 b791c2a7da..c4d37d3259 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 @@ -3293,6 +3293,218 @@ public void call() throws Exception } } + /** + * Test case for [Issue 954]: a change parked as waiting for another one is given back + * when the replay which parked it is unwound. + *

+ * A change which waits for another one is parked and stays owned by the replay thread + * which parked it, while that thread goes on to the changes which follow: it is handed + * out again by {@code getNextUpdate()}, which every replay loop of this domain runs once + * it is done, so it is replayed by whichever thread clears the change it was waiting for. + * A replay which is unwound leaves the thread which parked it without that road - it + * takes the next delivery off the shared queue instead, and never comes back to the + * change it parked - and every redelivery of a change a replay thread owns is refused as + * a duplicate. On a domain which then goes quiet that change is where this replica's + * ServerState, and every change behind it from every master, stops. + */ + @Test + public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception + { + testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain"); + logger.error(LocalizableMessage.raw( + "Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain")); + + final DN waitedOn = addEntryForChange("user.954.1"); + final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString(); + final DN other = addEntryForChange("user.954.2"); + final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + domain.resetUnreplayedChangeAlertThrottle(); + final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size"); + assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0, + "no change of this domain is waiting for another one when this test starts"); + + final CSNGenerator gen = new CSNGenerator(24, TimeThread.getTime()); + final CSN failing = gen.newCSN(); + final CSN parked = gen.newCSN(); + final CSN unwound = gen.newCSN(); + final List failingMods = generatemods("description", "the replay of this change fails"); + final String parkedDescription = "the change which was parked as a dependency"; + final List parkedMods = generatemods("description", parkedDescription); + final List unwoundMods = + generatemods("description", "the replay of this change is unwound"); + + /* + * The change whose replay fails is the barrier the parked change waits behind, so its + * budget must not be spent while this test is setting up: it is shortened once the + * change which was parked has been given back, and put back in the finally below. + */ + setReplayGiveUpDelay("unlimited"); + /* + * One replay thread, so that the change which is parked and the replay which is + * unwound after it are the same thread's: a parked change is left owned by the thread + * which parked it, and this is about a thread which does not come back to it. + */ + setNumUpdateReplayThreads(1); + try + { + /* + * A change whose replay failed stays listed and uncommitted - it is what holds this + * domain's ServerState back - and stays among the changes the newer ones are checked + * against, so a change which follows it on the same entry has to wait for it. It has + * to be a change which is asked for again rather than stepped over: one whose + * operation is built and then refused, since #928 has a modify whose entry DN does + * not parse reported once and recorded as replayed. + */ + deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1, + () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID), + "the change whose replay fails must stay listed as one which is not in the data"); + + // The change which is parked as waiting for it by the replay thread it was given to. + deliverUntilMonitorReaches(domain, "dependent-changes-size", 1, + () -> new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID), + "a change which waits for one that is not in the data must be parked"); + + /* + * The replay which is unwound while that same thread still holds the parked change. + * It is a change whose replay is unwound once the ack of its delivery is out: what + * the replay runs from there - the give-back of the change it was replaying, and the + * hand-out of the changes which were waiting for that one - is past every catch the + * replay itself has, so a throw there is what leaves replay() by the way this issue + * is about. An Error met replaying a change does not, and neither does a throw from + * publishing the ack: both are reported and take the ordinary road of a failed replay + * (issue #922). + * + * It is made on another entry, so that it is replayed rather than parked in its turn, + * and it is waited for on the changes being replayed rather than on the failure it + * reports: what this test is about happens on the way out of that replay. + */ + deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 3, + () -> new ModifyMsgWhoseReplayIsUnwoundAfterItsAck(unwound, other, unwoundMods, otherUUID), + "the replay of the change which follows the parked one must have been reached"); + + assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 0, + "a change parked by a replay which was unwound must be given back"); + + /* + * Nothing sends these changes again - they never travelled a session - so the + * deliveries which take over from the ones which were unwound are made here. The + * change no delivery can replay is given up on, which lets the ServerState past it, + * and the two which never failed are applied. + */ + setReplayGiveUpDelay(TEST_GIVE_UP_DELAY); + TestTimer timer = new TestTimer.Builder() + .maxSleep(120, SECONDS) + .sleepTimes(500, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + final ServerState state = domain.getServerState(); + if (!state.cover(failing)) + { + domain.processUpdate( + new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID)); + } + if (!state.cover(parked)) + { + domain.processUpdate(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID)); + } + if (!state.cover(unwound)) + { + domain.processUpdate(new ModifyMsg(unwound, other, unwoundMods, otherUUID)); + } + assertTrue(state.cover(parked), + "the change which was given back must be replayed by the delivery which takes it over"); + assertTrue(state.cover(unwound), + "the change whose replay was unwound must be replayed by the delivery which takes it over"); + } + }); + checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30, + "the change which was parked must be applied by the delivery which took it over"); + } + finally + { + resetReplayGiveUpDelay(); + resetNumUpdateReplayThreads(); + } + } + + /** + * Delivers a change until a monitor attribute of the domain reaches the expected value. + *

+ * A delivery is dropped rather than queued while the listener thread is down, which it + * is for as long as a recovery is restarting the session, so a change which has to reach + * a replay thread is delivered until it does. A delivery of a change a replay thread + * owns is refused as the duplicate it is, so the deliveries which follow the one that + * was taken cost nothing. + */ + private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain, + final String attributeName, final long expected, + final Supplier delivery, final String message) throws Exception + { + TestTimer timer = new TestTimer.Builder() + .maxSleep(20, SECONDS) + .sleepTimes(500, MILLISECONDS) + .toTimer(); + timer.repeatUntilSuccess(new CallableVoid() + { + @Override + public void call() throws Exception + { + if (getMonitorAttrValue(baseDN, attributeName) != expected) + { + domain.processUpdate(delivery.get()); + } + assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message); + } + }); + } + + /** + * Sets how many replay threads this server runs, the way an administrator would: the + * pool is stopped and created again with that number. + */ + private static void setNumUpdateReplayThreads(int threads) throws Exception + { + assertEquals(TestCaseUtils.applyModifications(true, + "dn: " + SYNCHRO_PLUGIN_DN, + "changetype: modify", + "replace: ds-cfg-num-update-replay-threads", + "ds-cfg-num-update-replay-threads: " + threads), 0, + "the number of replay threads could not be changed"); + } + + /** + * Puts the number of replay threads back to what this server computes for itself, which + * is what it runs with when the configuration carries no number of its own. + */ + private static void resetNumUpdateReplayThreads() throws Exception + { + assertEquals(TestCaseUtils.applyModifications(true, + "dn: " + SYNCHRO_PLUGIN_DN, + "changetype: modify", + "delete: ds-cfg-num-update-replay-threads"), 0, + "the number of replay threads could not be put back"); + } + + /** Adds the entry a change of these tests is made on. */ + private DN addEntryForChange(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").getName(); + } + /** A delivery of a change whose replay does not run to its end. */ private interface FailingDelivery { @@ -3313,16 +3525,7 @@ private interface FailingDelivery private void assertChangeIsDeliveredAgainAfter( FailingDelivery delivery, int serverId, String uid, final String description) throws Exception { - Entry tmp = TestCaseUtils.addEntry( - "dn: uid=" + uid + "," + baseDN, - "objectClass: top", - "objectClass: person", - "objectClass: organizationalPerson", - "objectClass: inetOrgPerson", - "uid: " + uid, - "cn: Aaccf Amar", - "sn: Amar"); - final DN dn = tmp.getName(); + final DN dn = addEntryForChange(uid); final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString(); final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java index e91292f43e..30107ac3b0 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java @@ -15,6 +15,7 @@ */ package org.opends.server.replication.plugin; +import static java.util.Collections.*; import static org.testng.Assert.*; import java.util.NoSuchElementException; @@ -670,10 +671,12 @@ public void run() /** * The change a thread parked as waiting for another one is not the change it is - * replaying: it is handed to whichever thread clears what it waits for, so a give-back on - * the way out of an unwound replay must leave it alone. Releasing it without taking it out - * of the changes which are waiting would have the same change handed to two threads - * (issue #922). + * replaying: it is handed to whichever thread clears what it waits for, so the give-back + * of the change a replay was unwound on must leave it alone. Releasing it without taking + * it out of the changes which are waiting would have the same change handed to two + * threads (issue #922) - which is why the parked ones are given back on a road of their + * own, {@link RemotePendingChanges#releaseParkedChangesOwnedByCurrentThread()}, where + * both happen in one step (issue #954). *

* The deliveries are taken in the order a replay thread takes them: one at a time, off * the queue the pool shares. So the change which is parked here is parked by the thread @@ -799,6 +802,145 @@ public void aChangeWhichWasWaitingIsHandedOutOnce() throws Exception "a change which has been handed out must not be handed out again"); } + /** + * Test case for [Issue 954]: a change parked as waiting for another one is given back + * when the replay which parked it is unwound. + *

+ * A parked change stays owned by the thread which parked it, and is handed out again by + * {@link RemotePendingChanges#getNextUpdate()} to whichever thread clears the change it + * waits for. A replay which is unwound - a JVM out of memory, a throw from what the replay + * runs once the ack of its delivery is out - leaves that thread without a road back to the + * change: it takes the next delivery off the queue instead. Nothing else asks for the + * change either, since every redelivery of a change a replay thread owns is refused as a + * duplicate, so this domain's ServerState would stay behind it until some other change is + * replayed on this domain. + */ + @Test + public void aChangeParkedByAReplayWhichIsUnwoundIsGivenBack() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + // A rename into the DN that delete is on: it can only be replayed once the delete has been. + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + + // The replay which parked it is unwound, so it gives back what it still owns. + assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), + singletonList(renamed), "the change this thread parked must be given back"); + + assertEquals(pendingChanges.getDependentChangesSize(), 0, + "a change which was given back must not be left waiting for a thread to hand it out"); + assertEquals(pendingChanges.getQueueSize(), 2, "the change must stay listed as pending"); + assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed"); + assertEquals(pendingChanges.changesInProgressSize(), 2, + "a change which is not in the data yet must stay a dependency of the changes which follow it"); + + assertTrue(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)), + "the change the unwound replay gave back must be taken over by the next delivery"); + } + + /** + * Test case for [Issue 954]: a change which was given back is handed out by the next + * delivery of it and by nothing else. + *

+ * Unparking it and giving it back is one step under both locks for that reason: a + * change which was released while it is still listed as waiting would be replayed by + * the thread {@link RemotePendingChanges#getNextUpdate()} hands it to and by the thread + * which takes over the delivery which follows - the double replay the ownership is + * there to prevent (OPENDJ-1115). + */ + @Test + public void aParkedChangeWhichWasGivenBackIsNotHandedOutAsADependency() throws Exception + { + final ServerState state = new ServerState(); + final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + pendingChanges.releaseParkedChangesOwnedByCurrentThread(); + + // The change it was waiting for is replayed, which is what used to hand it out. + pendingChanges.commit(deleted); + + assertNull(pendingChanges.getNextUpdate(), + "a change which was given back must not also be handed out as a dependency"); + + // It is replayed by the delivery which takes it over, and by that one only. + final ModifyDNMsg nextDelivery = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(nextDelivery)); + assertTrue(pendingChanges.markInProgress(nextDelivery)); + pendingChanges.commit(renamed); + + assertTrue(state.cover(renamed)); + assertEquals(pendingChanges.getQueueSize(), 0); + } + + /** + * Test case for [Issue 954]: the changes another thread parked are left alone. + *

+ * A change is given back by the thread which owns it and by nobody else, here as + * everywhere else (issue #922). The thread which parked a change is the one which may + * still be inside the dependency checks which parked it - they park a change once per + * dependency it has - so a change released under it would be listed as waiting again a + * moment later, and handed out while the delivery which took it over is being replayed. + */ + @Test + public void theParkedChangesOfAnotherThreadAreLeftAlone() throws Exception + { + final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState()); + final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); + final CSN deleted = generator.newCSN(); + final CSN renamed = generator.newCSN(); + + final DeleteMsg delete = deleteMsg(deleted, "uuid-1"); + assertTrue(pendingChanges.putRemoteUpdate(delete)); + assertTrue(pendingChanges.markInProgress(delete)); + + final ModifyDNMsg rename = renameIntoDeletedEntry(renamed); + assertTrue(pendingChanges.putRemoteUpdate(rename)); + assertTrue(pendingChanges.markInProgress(rename)); + assertTrue(pendingChanges.checkDependencies(rename)); + + // Another replay thread is unwound while this one holds the change it parked. + runAndJoin(new Runnable() + { + @Override + public void run() + { + assertEquals(pendingChanges.releaseParkedChangesOwnedByCurrentThread(), emptyList(), + "a change another thread parked is not this one's to give back"); + } + }); + + assertEquals(pendingChanges.getDependentChangesSize(), 1, + "the change must stay listed as waiting for the one it depends on"); + assertFalse(pendingChanges.putRemoteUpdate(renameIntoDeletedEntry(renamed)), + "a change a replay thread owns must not be taken over (OPENDJ-1115)"); + + // It is still handed to whichever thread clears the change it was waiting for. + pendingChanges.commit(deleted); + assertSame(pendingChanges.getNextUpdate(), rename); + } + /** A rename of an entry into the DN {@code deleteMsg(csn, "uuid-1")} deletes. */ private ModifyDNMsg renameIntoDeletedEntry(CSN csn) throws Exception {