From e8c0d1b865426420b4bd9c2e482b27be0e4df57f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Sat, 11 Jul 2026 10:37:27 +0000
Subject: [PATCH] [#710] Fix replication catch-up re-sending updates with the original assured flag (#714)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java | 27 +++++++++++++++++++++++++++
1 files changed, 27 insertions(+), 0 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 6e280e8..b6e1b2a 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
@@ -80,6 +80,13 @@
private final int maxQueueBytesSize;
/** Specifies whether the consumer is following the producer (is not late). */
private boolean following;
+ /**
+ * Specifies whether the last update message returned by
+ * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
+ * path) rather than taken from the in-memory {@link #msgQueue}. Only ever
+ * accessed by the single consumer thread calling {@link #getNextMessage()}.
+ */
+ private boolean lastMessageFromLateQueue;
/** Specifies the current serverState of this handler. */
private ServerState serverState;
/** Specifies the baseDN of the domain. */
@@ -226,6 +233,21 @@
}
/**
+ * Indicates whether the last update message returned by
+ * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
+ * path) rather than taken from the in-memory queue.
+ * <p>
+ * Must only be called from the consumer thread calling
+ * {@link #getNextMessage()}.
+ *
+ * @return true if the last returned update message came from the late queue
+ */
+ protected boolean isLastMessageFromLateQueue()
+ {
+ return lastMessageFromLateQueue;
+ }
+
+ /**
* Retrieves the name of this monitor provider. It should be unique among all
* monitor providers, including all instances of the same monitor provider.
*
@@ -319,6 +341,9 @@
msgQueue.consumeUpTo(msg);
if (updateServerState(msg))
{
+ // the returned instance is the one re-read from the
+ // changelog DB, not its msgQueue equivalent
+ lastMessageFromLateQueue = true;
return msg;
}
}
@@ -347,6 +372,7 @@
}
if (updateServerState(msg))
{
+ lastMessageFromLateQueue = true;
return msg;
}
continue;
@@ -379,6 +405,7 @@
* by the other server.
* Otherwise just loop to select the next message.
*/
+ lastMessageFromLateQueue = false;
return msg;
}
}
--
Gitblit v1.10.0