+ * A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on + * the wire, and its shutdown waits for that forward, so the announcement has to be in place + * before the message is published: one made afterwards is one the forward found nothing to + * clear, and the shutdown spends its whole grace period on a message which has already gone + * out. The broker may still refuse the message once it is announced, and then the + * announcement is withdrawn: what stays announced is what the broker reports as written. + */ + interface ReplicaOfflineAnnouncer + { + /** + * Announces that the replica goes offline at the provided CSN. + * + * @param offlineCSN + * the CSN of the ReplicaOfflineMsg which is about to be published + */ + void announce(CSN offlineCSN); + + /** + * Withdraws the announcement of a message the broker refused: it was written to no session, + * so nobody will forward it. + * + * @param offlineCSN + * the CSN of the ReplicaOfflineMsg which was announced and not published + */ + void withdraw(CSN offlineCSN); + } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java new file mode 100644 index 0000000000..bb6a162416 --- /dev/null +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java @@ -0,0 +1,62 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.plugin.PendingChanges.ReplicaOfflineAnnouncer; +import org.opends.server.replication.service.DSRSShutdownSync; + +/** + * Announces the replica of one domain offline to the {@link DSRSShutdownSync} the shutdown of a + * collocated replication server waits on, and takes such an announcement back. + *
+ * This is the announcer {@link LDAPReplicationDomain} hands its {@link PendingChanges}: the + * announcement goes through {@link DSRSShutdownSync#replicaOfflineMsgSent(DN, CSN)} and the + * withdrawal through {@link DSRSShutdownSync#replicaOfflineMsgNotSent(DN, CSN)}, for the domain + * the announcer was built for. + */ +final class ShutdownSyncAnnouncer implements ReplicaOfflineAnnouncer +{ + private final DSRSShutdownSync shutdownSync; + private final DN baseDN; + + /** + * Creates an announcer for the replica of one domain. + * + * @param shutdownSync + * the synchronization object the collocated replication server's shutdown waits on + * @param baseDN + * the domain whose replica announces itself + */ + ShutdownSyncAnnouncer(DSRSShutdownSync shutdownSync, DN baseDN) + { + this.shutdownSync = shutdownSync; + this.baseDN = baseDN; + } + + @Override + public void announce(CSN offlineCSN) + { + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + } + + @Override + public void withdraw(CSN offlineCSN) + { + shutdownSync.replicaOfflineMsgNotSent(baseDN, offlineCSN); + } +} diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java index ba432c4b3c..21e5aee36d 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java @@ -46,7 +46,7 @@ public class DSRSShutdownSync { /** * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated - * RS, in milliseconds, counted from the moment the message was sent. + * RS, in milliseconds, counted from the moment the message was announced. */ public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000; @@ -97,19 +97,90 @@ public DSRSShutdownSync() } /** - * Message has been sent. + * Message is about to be sent. + *
+ * The announcement comes before the message is published rather than after: a collocated + * replication server can forward the message as soon as it is on the wire, and a forward which + * finds nothing announced has nothing to clear. The announcement of a message the broker then + * refuses is taken back by {@link #replicaOfflineMsgNotSent(DN, CSN)}. + *
+ * A replica announces itself offline on every disableService(), so this may take the place of
+ * an earlier announcement of the same replica which is still owed its forward. The earlier one
+ * is kept behind the new one: a forward of the newer message, which the replication server
+ * queued behind the earlier one, covers both, and a withdrawal of the newer one gives the
+ * earlier one its wait back. It is kept only while its own grace period runs: past it, the
+ * announcement holds nothing back any more, and keeping it would chain every announcement of
+ * a replica whose message nobody in this process forwards - a directory server without a
+ * collocated replication server, or connected to a remote one - for the life of the process.
*
* @param baseDN
- * the domain for which the message has been sent
+ * the domain for which the message is being sent
* @param offlineCSN
- * the CSN of the message, which identifies both the replica which announced itself
+ * the CSN of the message, which identifies both the replica which announces itself
* offline and the announcement being waited for
*/
public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN)
{
+ final long announcedAt = System.nanoTime();
replicaOfflineMsgs
.computeIfAbsent(baseDN, dn -> new ConcurrentHashMap
+ * The announcement is made before the message is published, since a collocated replication
+ * server can forward it as soon as it is on the wire, so the announcement of a message the
+ * broker then refused has to be taken back: nobody will forward it, and the shutdown would
+ * spend the whole grace period waiting for that forward. Only the announcement carrying that
+ * CSN is withdrawn, and the announcement it displaced - an earlier message of the same replica
+ * which did go out and is still owed its forward - takes its place again.
+ *
+ * Whatever is reported about that earlier message while the announcement of the refused one
+ * stands in its place is not seen by it. A forward, or the loss of a peer it was queued for,
+ * is lost, and the shutdown then waits out what is left of the earlier message's own grace
+ * period; the peers it is queued for, if they are recorded in that window, are lost too, with
+ * the opposite effect - the first forward ends its wait, as for a message no peer was recorded
+ * for. That window is the one publish the broker refuses: at once on a connection error or a
+ * pending recovery, the broker's retry loop up to the reconnect when it has no session. The
+ * wait it can cost is bounded by a grace period which is already running.
+ *
+ * @param baseDN
+ * the domain for which the message was announced
+ * @param offlineCSN
+ * the CSN of the message which was not sent
+ */
+ public void replicaOfflineMsgNotSent(DN baseDN, CSN offlineCSN)
+ {
+ final ConcurrentMap
@@ -356,16 +436,23 @@ private static final class PendingOfflineMsg
private final CSN csn;
/** When the message was announced, on the {@link System#nanoTime()} clock. */
private final long sentTime;
+ /**
+ * The announcement of the same replica this one took the place of and which is still owed its
+ * forward, null when there was none or when its grace period had already expired. It is
+ * given its place back if this message is withdrawn.
+ */
+ private final PendingOfflineMsg displaced;
/**
* The replication servers the message was queued for and which have not forwarded it yet,
* null as long as it has not been queued for anybody.
*/
private volatile Set
- * These tests need no server: the changes are built by a CSNGenerator, which reads the time
- * service, and the time service is up as soon as its class is loaded.
+ * These tests need no server: the changes are built by a CSNGenerator, which needs nothing but
+ * a server id.
*/
@SuppressWarnings("javadoc")
@Test(groups = { "precommit", "replication" }, sequential = true)
public class PendingChangesTest extends DirectoryServerTestCase
{
private static final int SERVER_ID = 42;
+ /** A peer replication server the collocated one relays the message to. */
+ private static final int RS_ID = 11;
+
+ private static DN baseDN;
+
+ @BeforeClass
+ public static void classSetup() throws Exception
+ {
+ baseDN = DN.valueOf("dc=example,dc=com");
+ }
@Test
public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Exception
@@ -61,6 +76,47 @@ public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Excepti
assertEquals(published.getCSN(), offlineCSN);
}
+ /**
+ * The collocated replication server forwards the message as soon as it is on the wire, so an
+ * announcement made after the publish is one the forward found nothing to clear: nothing will
+ * ever remove it, and the shutdown waits out its whole grace period for a message which the
+ * topology already has.
+ *
+ * The forward is reported from inside publish(), which is where the message reaches the
+ * session, so the race is reproduced rather than waited for.
+ */
+ @Test
+ public void theReplicaOfflineMsgIsAnnouncedBeforeItIsPublished() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
+ final ReplicationDomain domain = mock(ReplicationDomain.class);
+ forwardWhilePublishing(domain, shutdownSync);
+ final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);
+
+ pendingChanges.putReplicaOfflineMsg();
+
+ assertTrue(shutdownSync.canShutdown(baseDN),
+ "the message was forwarded, so nothing must hold the shutdown back any longer");
+ }
+
+ /**
+ * The announcement of a message the broker took stands until a peer forwards it: the
+ * withdrawal is for the message the broker refused, and an announcement taken back after a
+ * publish which succeeded would leave the shutdown nothing to wait for.
+ */
+ @Test
+ public void theAnnouncementOfAPublishedMessageStandsUntilItIsForwarded() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
+ final PendingChanges pendingChanges =
+ newPendingChanges(domainWhichPublishes(true), shutdownSync);
+
+ pendingChanges.putReplicaOfflineMsg();
+
+ assertFalse(shutdownSync.canShutdown(baseDN),
+ "the message went out and nobody has forwarded it yet, so the shutdown must wait for it");
+ }
+
/**
* The broker writes nothing when it has no usable session, when the changes which come before
* this one still have to be republished by the recovery, or when it is stopped in between - and
@@ -79,6 +135,27 @@ public void replicaOfflineMsgTheBrokerRefusedIsNotReportedAsSent() throws Except
assertTrue(onlyMsgPublishedBy(domain) instanceof ReplicaOfflineMsg, "it was attempted");
}
+ /**
+ * The announcement is made before the message is published, and the broker may refuse it
+ * once it is: an announcement which stayed would be one nobody will ever forward, and the
+ * shutdown would wait out its whole grace period for a message which never left. It is
+ * therefore withdrawn - and it is a withdrawal, not an announcement which was never made: the
+ * shutdown is held back while the broker holds the message.
+ */
+ @Test
+ public void theAnnouncementOfAReplicaOfflineMsgTheBrokerRefusedIsWithdrawn() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
+ final ReplicationDomain domain = mock(ReplicationDomain.class);
+ refuseWhilePublishing(domain, shutdownSync);
+ final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);
+
+ assertNull(pendingChanges.putReplicaOfflineMsg(), "the broker refused the message");
+
+ assertTrue(shutdownSync.canShutdown(baseDN),
+ "the message never reached the wire, so nothing must hold the shutdown back");
+ }
+
/**
* The message carries the newest CSN of the replica, so a change which is still in flight
* holds it back, and the broker is never even asked to publish it.
@@ -116,6 +193,32 @@ public void replicaOfflineMsgWhichCouldNotBeSentIsNotPublishedLater() throws Exc
assertTrue(published instanceof LDAPUpdateMsg, "published " + published);
}
+ /**
+ * The announcement follows the publication rather than the queueing, so a message which a
+ * change in flight holds back is not announced: neither while it waits, nor when the change
+ * which held it back completes and the message is given up on. Announcing it either time
+ * would leave the shutdown waiting out its whole grace period for a forward which cannot
+ * happen.
+ */
+ @Test
+ public void theReplicaOfflineMsgHeldBackByAChangeInFlightIsNeverAnnounced() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
+ final ReplicationDomain domain = domainWhichPublishes(true);
+ final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);
+ final CSN inFlight = pendingChanges.putLocalOperation(newLocalOperation());
+
+ pendingChanges.putReplicaOfflineMsg();
+
+ assertTrue(shutdownSync.canShutdown(baseDN),
+ "the message is still queued behind a change in flight, and nothing was announced");
+
+ pendingChanges.commitAndPushCommittedChanges(inFlight, mock(LDAPUpdateMsg.class));
+
+ assertTrue(shutdownSync.canShutdown(baseDN),
+ "the message was given up on with the change which held it back, and never announced");
+ }
+
/**
* A change the broker refused leaves the pending changes all the same: the replica has done
* it, its ServerState says so, and it is by finding that state ahead of the one its
@@ -137,9 +240,55 @@ public void changeTheBrokerRefusedStillLeavesThePendingChanges() throws Exceptio
assertEquals(pendingChanges.size(), 0, "and is not queued for a second attempt");
}
+ /**
+ * Reports the forward of the message from within the publish which puts it on the wire, after
+ * checking from there that the announcement is already in place, and publishes it. The forward
+ * is the one of a peer the message was never recorded as queued for, which is what a forward
+ * racing the announcement looks like.
+ */
+ private void forwardWhilePublishing(
+ final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
+ {
+ doAnswer(invocation -> {
+ final UpdateMsg msg = (UpdateMsg) invocation.getArguments()[0];
+ if (msg instanceof ReplicaOfflineMsg)
+ {
+ assertFalse(shutdownSync.canShutdown(baseDN),
+ "the message must be announced before it is published");
+ shutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), RS_ID);
+ }
+ return true;
+ }).when(domain).publish(any(UpdateMsg.class));
+ }
+
+ /**
+ * Refuses to publish the message, the way a broker with no usable session does, after checking
+ * from within the publish that the announcement is already in place.
+ */
+ private void refuseWhilePublishing(
+ final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
+ {
+ doAnswer(invocation -> {
+ assertFalse(shutdownSync.canShutdown(baseDN),
+ "the message must be announced before it is published");
+ return false;
+ }).when(domain).publish(any(UpdateMsg.class));
+ }
+
private PendingChanges newPendingChanges(ReplicationDomain domain)
{
- return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain);
+ return newPendingChanges(domain, new DSRSShutdownSync());
+ }
+
+ /**
+ * The pending changes of a replica whose domain announces itself through the shutdown sync,
+ * with the announcer LDAPReplicationDomain hands its own pending changes.
+ */
+ private PendingChanges newPendingChanges(
+ final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
+ {
+ return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain,
+ new ShutdownSyncAnnouncer(shutdownSync, baseDN));
}
/** A domain whose broker accepts, or refuses, whatever it is given to publish. */
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java
index be482e18d0..a0479cc62a 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java
@@ -86,6 +86,150 @@ public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception
assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
}
+ /**
+ * The announcement is made before the message is published, so the broker may still refuse
+ * it - no usable session, or stopped in between. What was announced and never written must
+ * not hold the shutdown back: nobody will forward it.
+ */
+ @Test
+ public void canShutdownOnceTheReplicaOfflineMsgIsWithdrawn() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, offlineCSN);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * A withdrawal takes back the very announcement it names, and not whatever the replica
+ * announced last: a stale one, of a message the replica has since announced again, is ignored,
+ * and the newer announcement is still owed its forward. Ignored, and not taken for the
+ * withdrawal of the newer one: that would put the stale announcement back in front, where the
+ * forward of its message - which says nothing about the newer one - would end the wait.
+ */
+ @Test
+ public void theWithdrawalOfAnEarlierMessageLeavesANewerOneAlone() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN refusedByTheBroker = newCSN(SERVER_ID, 1);
+ final CSN sentByTheShutdown = newCSN(SERVER_ID, 2);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
+ shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
+
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, refusedByTheBroker, RS_ID);
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the stale withdrawal was ignored, not turned into a restore")
+ .isFalse();
+ }
+
+ /**
+ * A replica announces itself offline on every disableService(), and a later announcement
+ * takes the place of the earlier one. When the broker then refuses the later message, the
+ * earlier one - which did go out, and which a peer still has to forward - must get its wait
+ * back: withdrawing the later announcement must not take the earlier one with it.
+ */
+ @Test
+ public void theWithdrawalOfALaterMessageGivesTheEarlierOneItsWaitBack() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
+ final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
+
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the earlier message went out and nobody has forwarded it yet")
+ .isFalse();
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the forward of the earlier message ends the wait")
+ .isTrue();
+ }
+
+ /**
+ * What the withdrawal gives back is the very announcement which was displaced, with the peers
+ * its message was queued for: the forward of one of them does not end a wait which is for
+ * several, as it would for a message which was announced again from scratch.
+ */
+ @Test
+ public void theRestoredAnnouncementIsStillOwedTheForwardsItWasQueuedFor() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
+ final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgDispatched(
+ baseDN1, sentByTheShutdown, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
+
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the restored message is still owed the other peer's forward")
+ .isFalse();
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, OTHER_RS_ID);
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * The restored announcement keeps its own clock: the shutdown waits out what is left of the
+ * earlier message's grace period, not a new one counted from the withdrawal.
+ */
+ @Test
+ public void theRestoredAnnouncementKeepsWhatIsLeftOfItsOwnGracePeriod() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);
+ final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
+ final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
+ Thread.sleep(GRACE_PERIOD - 200);
+ shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
+ // past the end of the earlier message's grace period, well short of a whole new one
+ Thread.sleep(250);
+
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the wait started over at the withdrawal")
+ .isTrue();
+ }
+
+ /**
+ * While the announcement of a refused message stands in the place of the earlier one, what is
+ * reported about the earlier message is not seen by it: a forward reported in that window is
+ * lost, and the restored announcement waits out what is left of its own grace period. The
+ * window is the one refused publish; this pins the trade-off, so that a change to it is made
+ * knowingly.
+ */
+ @Test
+ public void aForwardReportedWhileARefusedAnnouncementStoodIsNotSeen() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
+ final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, sentByTheShutdown, asList(RS_ID));
+ shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
+ // not seen: the announcement of the refused message stands in front
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
+
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("a forward reported while the refused announcement stood is not seen")
+ .isFalse();
+ }
+
@Test
public void canShutdownOnceTheGracePeriodExpired() throws Exception
{
@@ -430,6 +574,55 @@ public void theWaitEndsWhenTheLastPeerExpectedToForwardStops() throws Exception
.isLessThan(LONG_GRACE_PERIOD);
}
+ /**
+ * A withdrawal must wake the shutdown up as a forward does, and not leave it waiting for the
+ * forward of a message which never left.
+ */
+ @Test
+ public void theWaitEndsWhenTheMessageIsWithdrawn() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ final Thread withdrawer = newWithdrawerThread(shutdownSync, offlineCSN);
+
+ final long startTime = System.nanoTime();
+ withdrawer.start();
+ shutdownSync.awaitReplicaOfflineMsgsForwarded(
+ asList(baseDN1), shutdownSync.newShutdownDeadline());
+ final long elapsed = millisSince(startTime);
+ withdrawer.join();
+
+ assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY);
+ assertThat(elapsed)
+ .as("the withdrawal did not wake the wait up")
+ .isLessThan(LONG_GRACE_PERIOD);
+ assertThat(shutdownSync.canShutdown(baseDN1))
+ .as("the withdrawn message holds nothing back")
+ .isTrue();
+ }
+
+ /** Withdraws the announcement, as the broker refusing the message during the wait does. */
+ private Thread newWithdrawerThread(final DSRSShutdownSync shutdownSync, final CSN offlineCSN)
+ {
+ return new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ Thread.sleep(FORWARD_DELAY);
+ shutdownSync.replicaOfflineMsgNotSent(baseDN1, offlineCSN);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ }
+
/** Stops the peer the message was queued for, as a disconnection during the wait does. */
private Thread newPeerStopperThread(final DSRSShutdownSync shutdownSync, final int peerId)
{