From 2c7e382da347e11820463fcf977d885c2f40d764 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 10 Sep 2026 08:50:34 +0000
Subject: [PATCH] [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg (#947)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java | 150 ++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 144 insertions(+), 6 deletions(-)
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 3d962b0..ba432c4 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
@@ -20,6 +20,8 @@
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import java.util.Collection;
+import java.util.Map.Entry;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -51,9 +53,13 @@
private final long gracePeriod;
/**
- * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per
+ * The ReplicaOfflineMsg which is still owed a forward, per domain and per
* replica of that domain.
* <p>
+ * An entry lives until every replication server the message was queued for
+ * has forwarded it, so it legitimately holds a message some of them have
+ * already sent: what is pending is the forward, not the message.
+ * <p>
* It is kept per domain because a domain sends this message whenever its
* replication service is disabled - an online import, a restore, a
* configuration change - and not only when the process shuts down. A single
@@ -63,6 +69,10 @@
* It is kept per replica because the collocated RS relays the message of
* every replica connected to it, and the forward of another replica's
* message says nothing about this one.
+ * <p>
+ * Each entry knows the replication servers its message was queued for, because each of them is
+ * served by its own writer: the forward of one of them says nothing about the others, whose
+ * queue the shutdown is about to clear.
*/
private final ConcurrentMap<DN, ConcurrentMap<Integer, PendingOfflineMsg>> replicaOfflineMsgs =
new ConcurrentHashMap<>();
@@ -103,14 +113,54 @@
}
/**
- * Message has been forwarded.
+ * Message has been queued for the replication servers which must forward it.
+ * <p>
+ * This must be called before the message is queued for any of them: a replication server can
+ * forward it as soon as it is in its queue, and a forward which finds no recipient recorded
+ * ends the wait at once.
+ *
+ * @param baseDN
+ * the domain for which the message has been sent
+ * @param offlineCSN
+ * the CSN of the message which is being queued
+ * @param replicationServerIds
+ * the server ids of the replication servers the message is being queued for
+ */
+ public void replicaOfflineMsgDispatched(
+ DN baseDN, CSN offlineCSN, Collection<Integer> replicationServerIds)
+ {
+ final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+ if (msgs == null)
+ {
+ return;
+ }
+ final int serverId = offlineCSN.getServerId();
+ final PendingOfflineMsg pending = msgs.get(serverId);
+ /*
+ * The message being queued may be an older announcement of the same replica - one which was
+ * queued behind a backlog since an earlier import. The replication servers it goes to say
+ * nothing about the announcement the shutdown is waiting for.
+ */
+ if (pending != null && pending.csn.equals(offlineCSN)
+ && pending.awaitForwardsFrom(replicationServerIds))
+ {
+ // queued for nobody: there is nothing to wait for
+ msgs.remove(serverId, pending);
+ notifyForwarded();
+ }
+ }
+
+ /**
+ * Message has been forwarded to one of the replication servers it was queued for.
*
* @param baseDN
* the domain for which the message has been sent
* @param forwardedCSN
* the CSN of the forwarded message
+ * @param replicationServerId
+ * the server id of the replication server the message has been forwarded to
*/
- public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN)
+ public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN, int replicationServerId)
{
final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
if (msgs != null)
@@ -124,11 +174,47 @@
* Such a forward says nothing about the announcement the shutdown is waiting for, and must
* not consume its grace period.
*/
- if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN))
+ if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN)
+ && pending.forwardedBy(replicationServerId))
{
msgs.remove(serverId, pending);
}
}
+ notifyForwarded();
+ }
+
+ /**
+ * A replication server the message may have been queued for will not forward it: it is gone,
+ * or the message was dropped on its way out.
+ * <p>
+ * Whatever it was given can no longer reach it, so the shutdown must not spend the rest of its
+ * grace period waiting for it.
+ *
+ * @param baseDN
+ * the domain the replication server is connected to
+ * @param replicationServerId
+ * the server id of the replication server which will not forward the message
+ */
+ public void replicaOfflineMsgNotForwarded(DN baseDN, int replicationServerId)
+ {
+ final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+ if (msgs != null)
+ {
+ for (Entry<Integer, PendingOfflineMsg> entry : msgs.entrySet())
+ {
+ final PendingOfflineMsg pending = entry.getValue();
+ if (pending.giveUpOn(replicationServerId))
+ {
+ msgs.remove(entry.getKey(), pending);
+ }
+ }
+ }
+ notifyForwarded();
+ }
+
+ /** Wakes up the shutdown, which re-reads what is left to wait for. */
+ private void notifyForwarded()
+ {
synchronized (forwardedMonitor)
{
forwardedMonitor.notifyAll();
@@ -137,7 +223,8 @@
/**
* Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg
- * has been forwarded or its grace period has expired.
+ * has been forwarded by every replication server it was queued for, or its
+ * grace period has expired.
* <p>
* The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)}
* rather than polling this; it is the same state, observable without waiting for it.
@@ -269,6 +356,11 @@
private final CSN csn;
/** When the message was announced, on the {@link System#nanoTime()} clock. */
private final long sentTime;
+ /**
+ * 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)
{
@@ -276,10 +368,56 @@
this.sentTime = sentTime;
}
+ /**
+ * Records the replication servers the message is being queued for, and returns whether there
+ * is none of them, i.e. nothing left to wait for.
+ */
+ private boolean awaitForwardsFrom(Collection<Integer> replicationServerIds)
+ {
+ final Set<Integer> awaited = ConcurrentHashMap.newKeySet();
+ awaited.addAll(replicationServerIds);
+ awaitedForwarders = awaited;
+ return awaited.isEmpty();
+ }
+
+ /**
+ * Records the forward of one replication server, and returns whether nothing is left to wait
+ * for.
+ */
+ private boolean forwardedBy(int replicationServerId)
+ {
+ final Set<Integer> awaited = awaitedForwarders;
+ if (awaited == null)
+ {
+ /*
+ * 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.
+ */
+ return true;
+ }
+ awaited.remove(replicationServerId);
+ return awaited.isEmpty();
+ }
+
+ /**
+ * Gives up on the forward of one replication server, and returns whether nothing is left to
+ * wait for. Unlike a forward, this releases nothing while no recipient is known: a peer going
+ * away says nothing about a message it was never given.
+ */
+ private boolean giveUpOn(int replicationServerId)
+ {
+ final Set<Integer> awaited = awaitedForwarders;
+ return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
+ }
+
@Override
public String toString()
{
- return "PendingOfflineMsg(" + csn + ")";
+ final Set<Integer> awaited = awaitedForwarders;
+ return "PendingOfflineMsg(" + csn
+ + (awaited != null ? ", awaiting the forward of " + awaited : "") + ")";
}
}
}
--
Gitblit v1.10.0