From 600df926522919b63bfca2816ef9588b6f1c6e34 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 15 Sep 2026 17:09:57 +0000
Subject: [PATCH] [#950] Announce a ReplicaOfflineMsg before it is published, not after it may have been forwarded (#978)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java | 62 ++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java | 55 +++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 15
opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java | 193 +++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java | 159 +++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java | 116 ++++++++++-
6 files changed, 574 insertions(+), 26 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 989d0e5..0b87409 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
@@ -828,7 +828,8 @@
* The generator time is adjusted to the time of the last CSN received from
* remote other servers.
*/
- pendingChanges = new PendingChanges(getGenerator(), this);
+ pendingChanges = new PendingChanges(getGenerator(), this,
+ new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN()));
remotePendingChanges = new RemotePendingChanges(getServerState());
// listen for changes on the configuration
@@ -2271,17 +2272,13 @@
public void publishReplicaOfflineMsg()
{
final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
- if (offlineCSN != null)
+ if (offlineCSN == null && logger.isTraceEnabled())
{
/*
- * Only a message which really was published is announced: the shutdown of a collocated
- * replication server waits for it to be forwarded, and would spend the whole grace
- * period waiting for one which never reached the wire.
+ * The announcement itself is made where the message is published, so nothing has to be
+ * reported here: a message a change in flight held back was never announced, and the
+ * announcement of one the broker refused has been withdrawn.
*/
- dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN);
- }
- else if (logger.isTraceEnabled())
- {
logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
+ " could not announce itself offline: the message was not published - a change which"
+ " is still in flight holds it back, or the broker had no session to write it to,"
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
index 4f88b62..231298f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
@@ -53,6 +53,9 @@
/** The ReplicationDomain that will be used to send UpdateMsg. */
private final ReplicationDomain domain;
+ /** Told that the replica of this domain announces itself offline, or takes that back. */
+ private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer;
+
private boolean recoveringOldChanges;
/**
@@ -60,11 +63,16 @@
*
* @param csnGenerator The CSNGenerator to use to create new unique CSNs.
* @param domain The ReplicationDomain that will be used to send UpdateMsg.
+ * @param replicaOfflineAnnouncer Told that the replica of this domain announces itself
+ * offline, before the message announcing it is published, and that it takes
+ * the announcement back when the broker refused the message.
*/
- PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain)
+ PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain,
+ ReplicaOfflineAnnouncer replicaOfflineAnnouncer)
{
this.csnGenerator = csnGenerator;
this.domain = domain;
+ this.replicaOfflineAnnouncer = replicaOfflineAnnouncer;
}
/**
@@ -198,9 +206,22 @@
}
else if (msg instanceof ReplicaOfflineMsg)
{
+ /*
+ * Announce the replica offline before the message reaches the wire, and not after:
+ * a collocated replication server forwards it as soon as it has it, and a forward
+ * which finds nothing announced leaves the shutdown waiting out the whole grace
+ * period of a message the topology already has.
+ */
+ final CSN offlineCSN = msg.getCSN();
+ replicaOfflineAnnouncer.announce(offlineCSN);
if (domain.publish(msg))
{
- publishedOfflineCSN = msg.getCSN();
+ publishedOfflineCSN = offlineCSN;
+ }
+ else
+ {
+ // The broker wrote it to no session, so nobody will forward what was announced.
+ replicaOfflineAnnouncer.withdraw(offlineCSN);
}
}
@@ -266,4 +287,34 @@
}
return recoveringOldChanges;
}
+
+ /**
+ * Told that the replica of this domain announces itself offline, or takes that back.
+ * <p>
+ * 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 0000000..bb6a162
--- /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.
+ * <p>
+ * 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 ba432c4..21e5aee 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 @@
{
/**
* 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 @@
}
/**
- * Message has been sent.
+ * Message is about to be sent.
+ * <p>
+ * 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)}.
+ * <p>
+ * 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<Integer, PendingOfflineMsg>())
- .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime()));
+ .compute(offlineCSN.getServerId(), (serverId, displaced) ->
+ new PendingOfflineMsg(offlineCSN, announcedAt,
+ displaced != null && gracePeriodLeft(displaced, announcedAt) > 0 ? displaced : null));
+ }
+
+ /**
+ * The message which was announced was not sent after all: the broker had no session to write
+ * it to, or was stopped before it could.
+ * <p>
+ * 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.
+ * <p>
+ * 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<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+ if (msgs != null)
+ {
+ final int serverId = offlineCSN.getServerId();
+ final PendingOfflineMsg pending = msgs.get(serverId);
+ if (pending != null && pending.csn.equals(offlineCSN))
+ {
+ /*
+ * The displaced announcement may owe nothing any more: the forward which released it
+ * can have been reported while this announcement was being made, so that its remove(),
+ * which matches the entry it read, found this one in its place. Given its place back,
+ * such an announcement would hold the shutdown for the rest of its grace period, since
+ * nobody will report that forward again.
+ */
+ if (pending.displaced != null && !pending.displaced.isFullyForwarded())
+ {
+ msgs.replace(serverId, pending, pending.displaced);
+ }
+ else
+ {
+ msgs.remove(serverId, pending);
+ }
+ }
+ }
+ notifyForwarded();
}
/**
@@ -338,12 +409,21 @@
long remaining = 0;
for (PendingOfflineMsg pending : msgs.values())
{
- remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime));
+ remaining = Math.max(remaining, gracePeriodLeft(pending, now));
}
return remaining;
}
/**
+ * Returns the time left, in milliseconds, of the grace period of one announcement, zero or
+ * less once it has expired.
+ */
+ private long gracePeriodLeft(PendingOfflineMsg pending, long now)
+ {
+ return gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime);
+ }
+
+ /**
* A ReplicaOfflineMsg a replica announced and which has not been forwarded yet.
* <p>
* This deliberately does not override {@code equals}: the two-argument
@@ -357,15 +437,22 @@
/** 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<Integer> awaitedForwarders;
- private PendingOfflineMsg(CSN csn, long sentTime)
+ private PendingOfflineMsg(CSN csn, long sentTime, PendingOfflineMsg displaced)
{
this.csn = csn;
this.sentTime = sentTime;
+ this.displaced = displaced;
}
/**
@@ -391,9 +478,8 @@
{
/*
* The message never went through the collocated RS - a replica which picked a remote one
- * announcing itself offline, or an announcement recorded after the message it belongs to
- * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait
- * had before the recipients were tracked: the first forward ends it.
+ * is announcing itself offline. Nobody is known to owe a forward, so keep the behaviour
+ * the wait had before the recipients were tracked: the first forward ends it.
*/
return true;
}
@@ -412,6 +498,16 @@
return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
}
+ /**
+ * Returns whether every replication server the message was queued for has forwarded it, or
+ * has been given up on: nothing is left to wait for. False while no recipient is known.
+ */
+ private boolean isFullyForwarded()
+ {
+ final Set<Integer> awaited = awaitedForwarders;
+ return awaited != null && awaited.isEmpty();
+ }
+
@Override
public String toString()
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
index 23c026f..1842e63 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
@@ -16,12 +16,14 @@
package org.opends.server.replication.plugin;
import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;
+import org.forgerock.opendj.ldap.DN;
import org.mockito.ArgumentCaptor;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
@@ -29,23 +31,36 @@
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.UpdateMsg;
+import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.types.operation.PluginOperation;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
/**
* Tests the bookkeeping a replica does on its own changes: they are published in the order of
- * their CSNs, and the announcement that the replica goes offline is only reported as sent when
- * it really was published.
+ * their CSNs, and the announcement that the replica goes offline is made before the message is
+ * published, since the shutdown of a collocated replication server waits for that message to be
+ * forwarded - and stands only for a message the broker reports as written.
* <p>
- * 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
@@ -62,6 +77,47 @@
}
/**
+ * 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.
+ * <p>
+ * 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
* what was not written must not be reported as sent: the shutdown of a collocated replication
@@ -80,6 +136,27 @@
}
/**
+ * 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.
*/
@@ -117,6 +194,32 @@
}
/**
+ * 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
* replication server reports that the next session republishes the change from the historical
@@ -137,9 +240,55 @@
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 be482e1..a0479cc 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 @@
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 @@
.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)
{
--
Gitblit v1.10.0