From 49bde76364c753aa8e1ddbca0581216b1939add2 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 16 Oct 2026 17:28:36 +0000
Subject: [PATCH] [#1198] Take the own state of a replication server again after each answer of its peers when it gathers the monitoring data (#1205)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java | 18 +++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorTest.java | 246 +++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 263 insertions(+), 1 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
index 48c3169..d92aab4 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server;
@@ -201,6 +202,8 @@
pendingMonitorData.completeComputing();
monitorData = pendingMonitorData;
monitorDataLastBuildDate = TimeThread.getTime();
+ // A late answer must not go into the published data: it would come after our own state.
+ pendingMonitorData = null;
}
}
finally
@@ -219,7 +222,17 @@
return monitorData;
}
- /** Start collecting global monitoring information for the replication domain. */
+ /**
+ * Takes the states of this RS and of its directly connected DSs into the pending monitor data:
+ * once when the computing starts, and again after each answer of another RS is stored.
+ * <p>
+ * The changes of our directly connected DSs reach the other RSs through us only, so a state of
+ * ours taken before theirs would show us behind them on those changes, and a DS reading that in
+ * the monitoring message would leave us for an RS which only has its changes from us (issue
+ * #1198). Taken again at each answer rather than once the wait for the answers is over, our
+ * state is no later than the last answer either: an RS which does not answer would otherwise put
+ * the whole wait between our state and the answers of the others, and they would look behind.
+ */
private void initializePendingMonitorData()
{
// Let's process our directly connected DS
@@ -318,6 +331,9 @@
}
}
}
+
+ // Our own state must not be older than the one this RS just answered with (issue #1198)
+ initializePendingMonitorData();
}
catch (RuntimeException e)
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorTest.java
new file mode 100644
index 0000000..698ded8
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorTest.java
@@ -0,0 +1,246 @@
+/*
+ * 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.server;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+import static org.testng.Assert.*;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.MonitorMsg;
+import org.opends.server.replication.protocol.ReplicationMsg;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Tests the states {@link ReplicationDomainMonitor} gathers for the replication servers of a
+ * domain, which the monitoring messages hand to the directory servers.
+ * <p>
+ * A directory server connected to a replication server checks on those messages whether its
+ * replication server still holds its newest change, and leaves it when another one holds it and
+ * it does not (issue #1198). The changes of a directory server reach the other replication
+ * servers through the one it is connected to only, so none of them can hold one of its changes
+ * which that one does not - unless the states compared were not taken at the same moment.
+ */
+@SuppressWarnings("javadoc")
+public class ReplicationDomainMonitorTest extends DirectoryServerTestCase
+{
+ private static final DN BASE_DN = DN.valueOf("dc=example,dc=com");
+ private static final int LOCAL_RS_ID = 1;
+ private static final int PEER_RS_ID = 2;
+ private static final int SILENT_RS_ID = 3;
+ private static final int DS_ID = 10;
+ private static final CSN FIRST = new CSN(1000, 1, DS_ID);
+ private static final CSN FORWARDED = new CSN(1000, 2, DS_ID);
+ private static final CSN AHEAD = new CSN(1000, 5, DS_ID);
+ private static final long STEP_TIMEOUT_MS = 10000;
+
+ /** What the reader of the session to a peer does once the monitor request has been sent to it. */
+ private interface PeerReader
+ {
+ void run(ReplicationDomainMonitor monitor, Thread requester) throws Exception;
+ }
+
+ /** The state of the changelog of this replication server, which the mocked domain reports. */
+ private AtomicReference<ServerState> changelog;
+ /** Counted down once the monitor data is published. */
+ private CountDownLatch published;
+ private Map<Integer, Thread> readers;
+ private AtomicReference<Throwable> readerFailure;
+
+ @BeforeMethod
+ public void setUp()
+ {
+ changelog = new AtomicReference<>(stateWith(FIRST));
+ published = new CountDownLatch(1);
+ readers = new ConcurrentHashMap<>();
+ readerFailure = new AtomicReference<>();
+ }
+
+ /**
+ * A change of a directory server connected to this replication server comes in while the
+ * answers are waited for, and this replication server forwards it to its peer before the peer
+ * answers: the state reported for this replication server must not be older than the one of its
+ * peer on the changes of that directory server.
+ */
+ @Test
+ public void aPeerDoesNotLookAheadOnTheChangesItGetsFromThisReplicationServer() throws Exception
+ {
+ final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
+ peers.put(PEER_RS_ID, (monitor, requester) ->
+ {
+ awaitTheWaitForTheAnswers(requester);
+ changelog.set(stateWith(FORWARDED));
+ monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, FORWARDED), PEER_RS_ID);
+ });
+
+ final ReplicationDomainMonitorData data = recompute(peers);
+
+ final CSN local = data.getRSStates(LOCAL_RS_ID).getCSN(DS_ID);
+ final CSN peer = stateOf(data, PEER_RS_ID).getCSN(DS_ID);
+ assertFalse(local.isOlderThan(peer), "RS(" + LOCAL_RS_ID + ") is reported at " + local.toStringUI()
+ + ", behind RS(" + PEER_RS_ID + ") at " + peer.toStringUI()
+ + ", which got the changes of DS(" + DS_ID + ") from it");
+ }
+
+ /**
+ * A peer can hold changes of a directory server which this replication server does not, when
+ * the directory server published them to that peer on an earlier session: the peer must still
+ * be reported ahead then.
+ */
+ @Test
+ public void aPeerWhichHoldsChangesThisReplicationServerLacksIsReportedAhead() throws Exception
+ {
+ final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
+ peers.put(PEER_RS_ID,
+ (monitor, requester) -> monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, AHEAD), PEER_RS_ID));
+
+ final ReplicationDomainMonitorData data = recompute(peers);
+
+ assertEquals(data.getRSStates(LOCAL_RS_ID).getCSN(DS_ID), FIRST);
+ assertEquals(stateOf(data, PEER_RS_ID).getCSN(DS_ID), AHEAD);
+ }
+
+ /**
+ * One peer answers at once, another one only once the wait for the answers is over, and changes
+ * come in meanwhile: the peer which answered is not shown missing the changes which came in
+ * while the other one was waited for, and the late answer does not go into the published data.
+ */
+ @Test
+ public void aPeerWhichAnswersAfterTheWaitLeavesTheOthersUpToDate() throws Exception
+ {
+ final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
+ peers.put(PEER_RS_ID,
+ (monitor, requester) -> monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, FIRST), PEER_RS_ID));
+ peers.put(SILENT_RS_ID, (monitor, requester) ->
+ {
+ awaitTheWaitForTheAnswers(requester);
+ readers.get(PEER_RS_ID).join(STEP_TIMEOUT_MS);
+ changelog.set(stateWith(FORWARDED));
+ assertTrue(published.await(STEP_TIMEOUT_MS, MILLISECONDS), "the monitor data was not published");
+ monitor.receiveMonitorDataResponse(answer(SILENT_RS_ID, FORWARDED), SILENT_RS_ID);
+ });
+
+ final ReplicationDomainMonitorData data = recompute(peers);
+
+ assertEquals(data.getMissingChangesRS(PEER_RS_ID), 0, "RS(" + PEER_RS_ID + ") answered with every "
+ + "change there was then, and is shown missing the ones which came in while RS(" + SILENT_RS_ID
+ + ") was waited for");
+ assertNull(data.getRSStates(SILENT_RS_ID),
+ "the answer of RS(" + SILENT_RS_ID + "), which came after the wait, went into the published data");
+ }
+
+ /**
+ * Gathers the monitor data of a domain with the provided peers, in this order: the reader of
+ * the session to each of them runs once the monitor request is sent to it. The readers are done
+ * when this returns.
+ */
+ private ReplicationDomainMonitorData recompute(Map<Integer, PeerReader> peerReaders) throws Exception
+ {
+ final ReplicationServerDomain domain = mock(ReplicationServerDomain.class);
+ when(domain.getBaseDN()).thenReturn(BASE_DN);
+ when(domain.getLocalRSServerId()).thenReturn(LOCAL_RS_ID);
+ when(domain.getConnectedDSs()).thenReturn(Collections.<Integer, DataServerHandler> emptyMap());
+ when(domain.getLatestServerState()).thenAnswer(invocation -> changelog.get().duplicate());
+ final ReplicationDomainMonitor monitor = new ReplicationDomainMonitor(domain);
+
+ final Map<Integer, ReplicationServerHandler> peers = new LinkedHashMap<>();
+ for (Map.Entry<Integer, PeerReader> entry : peerReaders.entrySet())
+ {
+ final int peerId = entry.getKey();
+ final PeerReader peerReader = entry.getValue();
+ final ReplicationServerHandler peer = mock(ReplicationServerHandler.class);
+ when(peer.getServerId()).thenReturn(peerId);
+ doAnswer(invocation ->
+ {
+ final Thread requester = Thread.currentThread();
+ final Thread reader = new Thread(() ->
+ {
+ try
+ {
+ peerReader.run(monitor, requester);
+ }
+ catch (Throwable e)
+ {
+ readerFailure.compareAndSet(null, e);
+ }
+ }, "reader of the session to RS(" + peerId + ")");
+ reader.setDaemon(true);
+ readers.put(peerId, reader);
+ reader.start();
+ return null;
+ }).when(peer).send(any(ReplicationMsg.class));
+ peers.put(peerId, peer);
+ }
+ when(domain.getConnectedRSs()).thenReturn(peers);
+
+ final ReplicationDomainMonitorData data = monitor.recomputeMonitorData();
+ published.countDown();
+ for (Map.Entry<Integer, Thread> reader : readers.entrySet())
+ {
+ reader.getValue().join(STEP_TIMEOUT_MS);
+ assertFalse(reader.getValue().isAlive(), "the reader of the session to RS(" + reader.getKey() + ") is stuck");
+ }
+ if (readerFailure.get() != null)
+ {
+ throw new AssertionError("a reader failed", readerFailure.get());
+ }
+ return data;
+ }
+
+ /** Returns once the requests are all sent, and the requester waits for the answers. */
+ private static void awaitTheWaitForTheAnswers(Thread requester) throws InterruptedException
+ {
+ final long deadline = System.currentTimeMillis() + STEP_TIMEOUT_MS;
+ while (requester.getState() != Thread.State.TIMED_WAITING)
+ {
+ assertTrue(System.currentTimeMillis() < deadline, "the requester did not wait for the answers");
+ Thread.sleep(1);
+ }
+ }
+
+ private static ServerState stateOf(ReplicationDomainMonitorData data, int rsId)
+ {
+ final ServerState state = data.getRSStates(rsId);
+ assertNotNull(state, "the answer of RS(" + rsId + ") did not come within the wait");
+ return state;
+ }
+
+ private static MonitorMsg answer(int rsId, CSN newest)
+ {
+ final MonitorMsg answer = new MonitorMsg(rsId, LOCAL_RS_ID);
+ answer.setReplServerDbState(stateWith(newest));
+ return answer;
+ }
+
+ private static ServerState stateWith(CSN csn)
+ {
+ final ServerState state = new ServerState();
+ state.update(csn);
+ return state;
+ }
+}
--
Gitblit v1.10.0