From b8a4d35f03b7fa5cf0d80a9424224f6ddeb3b3f9 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 19 Oct 2026 08:52:10 +0000
Subject: [PATCH] [#1206] Take a late handler of a replication server back to its queue only once its consumer has every change the queue lacks (#1219)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java | 95 ++++++++++++++++++++++++++++++++++++++++++++---
1 files changed, 89 insertions(+), 6 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
index 903b970..5d4596a 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
@@ -13,7 +13,7 @@
*
* Copyright 2009-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2026 3A Systems, LLC
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server;
@@ -81,6 +81,19 @@
/** Specifies whether the consumer is following the producer (is not late). */
private boolean following;
/**
+ * For each replica, the newest of its changes the queue may lack: the changes the domain held
+ * when it handed the queue its first change - the ones before never reached it - and the changes
+ * the queue dropped when it was full. Once the handler is registered, the domain hands it every
+ * change it receives which is meant for its consumer: a peer replication server is handed only
+ * the changes of the directory servers connected to this one, and a directory server nothing
+ * while its status is bad generation id or full update, which it leaves only with a new session.
+ * The consumer must have those the changelog still holds before it is served from the queue
+ * again, see {@link #consumerHasWhatTheQueueLacks(ServerState)}. Guarded by {@link #msgQueue}.
+ */
+ private final ServerState queueLacksUpTo = new ServerState();
+ /** Whether the queue has been handed a change yet. Guarded by {@link #msgQueue}. */
+ private boolean queueHandedAChange;
+ /**
* The state of the domain at the previous wait of {@link #getNextMessage()} on an empty queue,
* or {@code null} when the queue was not empty then. A change it holds has had a whole wait to
* reach the queue, see {@link #stopFollowingWhenChangesAreMissing()}. Guarded by
@@ -146,17 +159,67 @@
msgQueue.notifyAll();
}
+ if (!queueHandedAChange)
+ {
+ // the changes the domain held before this one never reached the queue. The domain writes
+ // a change to the changelog before it hands it over, so this one is counted as well
+ queueHandedAChange = true;
+ queueLacksUpTo.update(replicationServerDomain.getLatestServerState());
+ }
msgQueue.add(update);
// TODO : size should be configurable and larger than max-receive-queue-size
while (isMsgQueueAboveThreshold())
{
following = false;
- msgQueue.removeFirst();
+ queueLacksUpTo.update(msgQueue.removeFirst().getCSN());
}
}
}
+ /**
+ * Whether the consumer has every change the queue may lack which the changelog still holds - see
+ * {@link #queueLacksUpTo} and {@link #consumerLacksFromTheChangelog(CSN, ServerState)}.
+ * <p>
+ * Must be called while holding the {@link #msgQueue} monitor.
+ *
+ * @param inTheChangelog
+ * the newest change of each replica the changelog holds
+ * @return {@code true} when the queue lacks nothing the consumer may still be sent
+ */
+ private boolean consumerHasWhatTheQueueLacks(ServerState inTheChangelog)
+ {
+ for (CSN lacked : queueLacksUpTo)
+ {
+ if (consumerLacksFromTheChangelog(lacked, inTheChangelog))
+ {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Whether the consumer lacks the given change, and a read of the changelog can still give it the
+ * changes of that replica it lacks. A change the changelog no longer holds, such as one of a
+ * changelog cleared for a new generation id, cannot be read from it: once the consumer has every
+ * change of its replica the changelog holds, no read gives anything more of that replica, and such
+ * a change must not keep this handler from its queue.
+ *
+ * @param csn
+ * the CSN of the change
+ * @param inTheChangelog
+ * the newest change of each replica the changelog holds
+ * @return {@code true} when the consumer lacks the change and the changelog holds a change of its
+ * replica the consumer lacks
+ */
+ private boolean consumerLacksFromTheChangelog(CSN csn, ServerState inTheChangelog)
+ {
+ final CSN newest = inTheChangelog.getCSN(csn.getServerId());
+ final CSN sent = serverState.getCSN(csn.getServerId());
+ return !serverState.cover(csn) && newest != null && (sent == null || newest.isNewerThan(sent));
+ }
+
private boolean isMsgQueueAboveThreshold()
{
final long count = msgQueue.count();
@@ -339,8 +402,11 @@
synchronized (msgQueue)
{
// Ensure we are below threshold so this server will follow the
- // msgQueue without fearing the msgQueue gets trimmed
- if (isMsgQueueBelowThreshold())
+ // msgQueue without fearing the msgQueue gets trimmed.
+ // The changelog was read without the lock: a change written since may have been
+ // dropped from the queue already
+ if (isMsgQueueBelowThreshold()
+ && consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
{
following = true;
}
@@ -350,8 +416,19 @@
{
/*
* if the first change in the lateQueue is also on the regular queue,
+ * and the regular queue holds every other change the consumer lacks,
* we can resume the processing from the regular queue
* -> set following to true and empty the lateQueue.
+ * The regular queue is ordered by CSN, so holding the first change of the lateQueue does
+ * not mean holding the changes after it: a change which reaches the queue late with an
+ * older CSN - relayed by another replication server, or published again by a recovery -
+ * goes in front of the changes of another replica which the queue dropped before it came,
+ * or never had as they reached the domain before this handler did. Nor does it mean that
+ * the changes before it are ones the consumer has: one such late change may have reached
+ * the queue since the changelog was read. Served from the queue then, the consumer would
+ * not be sent those changes, and its state would move past them for good - see #1206.
+ * Only the changes the changelog can still give count: a read which gives nothing but
+ * replica offline messages would otherwise be made again and again, with no wait.
*/
UpdateMsg msg = lateQueue.first();
synchronized (msgQueue)
@@ -365,7 +442,10 @@
}
}
- if (msgQueue.contains(msg))
+ final ServerState inTheChangelog = replicationServerDomain.getLatestServerState();
+ if (msgQueue.contains(msg)
+ && consumerHasWhatTheQueueLacks(inTheChangelog)
+ && !msgQueue.holdsAChangeBefore(msg, csn -> consumerLacksFromTheChangelog(csn, inTheChangelog)))
{
/* we finally catch up with the regular queue */
following = true;
@@ -395,8 +475,11 @@
if (lateQueue.isEmpty() && msgQueue.isEmpty())
{
+ // no change left after the state of the consumer means it has what the queue lacks,
+ // unless the changelog could not be read
CSN nextChange = findOldestCSNFromReplicaDBs();
- if (nextChange == null)
+ if (nextChange == null
+ && consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
{
following = true;
}
--
Gitblit v1.10.0