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/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java | 868 +++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java | 9
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java | 7
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java | 16
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java | 10
opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java | 105 +++++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java | 24 +
opendj-server-legacy/src/main/java/org/opends/server/replication/server/MsgQueue.java | 27 +
opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java | 95 ++++
9 files changed, 1,141 insertions(+), 20 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;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MsgQueue.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MsgQueue.java
index 2564dcc..6d84ebd 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MsgQueue.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MsgQueue.java
@@ -20,6 +20,7 @@
import static org.opends.messages.ReplicationMessages.*;
import java.util.TreeMap;
+import java.util.function.Predicate;
import net.jcip.annotations.ThreadSafe;
@@ -178,6 +179,32 @@
}
}
+ /**
+ * Returns whether this queue holds, before the given UpdateMsg, an UpdateMsg which carries the
+ * CSN of a change and matches the given condition: a replica offline message, for one, does not
+ * carry the CSN of a change, and no state ever covers its CSN.
+ *
+ * @param msg
+ * the UpdateMsg the ones to check come before
+ * @param condition
+ * the condition the CSN of a change must match
+ * @return {@code true} if a change of this queue before the given UpdateMsg matches the condition
+ */
+ public boolean holdsAChangeBefore(UpdateMsg msg, Predicate<CSN> condition)
+ {
+ synchronized (lock)
+ {
+ for (UpdateMsg before : map.headMap(msg.getCSN()).values())
+ {
+ if (before.contributesToDomainState() && condition.test(before.getCSN()))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
/** Removes all UpdateMsg form this queue. */
public void clear()
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
index 8a91472..ca8e5ca 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
@@ -923,9 +923,12 @@
final Pair<FileReplicaDB, Boolean> pair = getOrCreateReplicaDB(baseDN,
csn.getServerId(), replicationServer);
final FileReplicaDB replicaDB = pair.getFirst();
- replicaDB.add(updateMsg);
-
- ChangelogBackend.getInstance().notifyCookieEntryAdded(baseDN, updateMsg);
+ if (replicaDB.add(updateMsg))
+ {
+ // a change the replica DB does not take - one it holds already, sent again - is not in the
+ // changelog a second time, and the persistent searches were told of it already
+ ChangelogBackend.getInstance().notifyCookieEntryAdded(baseDN, updateMsg);
+ }
final ChangeNumberIndexer indexer = cnIndexer.get();
if (indexer != null)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java
index 0dd267e..8fad4d0 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java
@@ -155,10 +155,12 @@
*
* @param updateMsg
* The update message to add.
+ * @return {@code true} if the message was added, {@code false} if it was not: this replica DB
+ * holds it already or holds a newer change, see {@link Log#append(Record)}
* @throws ChangelogException
* If an error occurs when trying to add the message.
*/
- void add(final UpdateMsg updateMsg) throws ChangelogException
+ boolean add(final UpdateMsg updateMsg) throws ChangelogException
{
if (shutdown.get())
{
@@ -167,7 +169,10 @@
.toString(), String.valueOf(baseDN), String.valueOf(serverId)));
}
- log.append(Record.from(updateMsg.getCSN(), updateMsg));
+ if (!log.append(Record.from(updateMsg.getCSN(), updateMsg)))
+ {
+ return false;
+ }
final CSNLimits limits = csnLimits;
final boolean updateNew = limits.newestCSN == null || limits.newestCSN.isOlderThan(updateMsg.getCSN());
@@ -178,6 +183,7 @@
updateOld ? updateMsg.getCSN() : limits.oldestCSN,
updateNew ? updateMsg.getCSN() : limits.newestCSN);
}
+ return true;
}
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
index 0839b7c..61d8233 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java
@@ -12,6 +12,7 @@
* information: "Portions Copyright [year] [name of copyright owner]".
*
* Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server.changelog.file;
@@ -411,10 +412,12 @@
*
* @param record
* The record to add.
+ * @return {@code true} if the record was appended, {@code false} if it was not: its key is not
+ * higher than the key of the last record added, or this log is closed
* @throws ChangelogException
* If an error occurs while adding the record to the log.
*/
- public void append(final Record<K, V> record) throws ChangelogException
+ public boolean append(final Record<K, V> record) throws ChangelogException
{
// Fast-path - assume that no rotation is needed and use shared lock.
sharedLock.lock();
@@ -422,13 +425,12 @@
{
if (isClosed)
{
- return;
+ return false;
}
LogFile<K, V> headLogFile = getHeadLogFile();
if (!mustRotate(headLogFile))
{
- headLogFile.append(record);
- return;
+ return headLogFile.append(record);
}
}
finally
@@ -442,13 +444,13 @@
{
if (isClosed)
{
- return;
+ return false;
}
LogFile<K, V> headLogFile = getHeadLogFile();
if (headLogFile.appendWouldBreakKeyOrdering(record))
{
// abort rotation
- return;
+ return false;
}
if (mustRotate(headLogFile))
{
@@ -457,7 +459,7 @@
rotateHeadLogFile();
headLogFile = getHeadLogFile();
}
- headLogFile.append(record);
+ return headLogFile.append(record);
}
finally
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
index a385f58..299ca53 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java
@@ -253,10 +253,12 @@
*
* @param record
* The record to add.
+ * @return {@code true} if the record was appended, {@code false} if its key is not higher than
+ * the key of the last record added
* @throws ChangelogException
* If the record can't be added to the log.
*/
- void append(final Record<K, V> record) throws ChangelogException
+ boolean append(final Record<K, V> record) throws ChangelogException
{
checkLogIsEnabledForWrite();
exclusiveLock.lock();
@@ -264,10 +266,11 @@
{
if (appendWouldBreakKeyOrdering(record))
{
- return;
+ return false;
}
writer.write(record);
newestRecord = record;
+ return true;
}
finally
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java
index e7be2ac..63014f0 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java
@@ -34,6 +34,7 @@
import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -43,6 +44,7 @@
import java.util.Set;
import java.util.SortedSet;
import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
import org.assertj.core.api.SoftAssertions;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -58,12 +60,16 @@
import org.opends.server.backends.ChangelogBackend.ChangeNumberRange;
import org.opends.server.controls.EntryChangelogNotificationControl;
import org.opends.server.controls.ExternalChangelogRequestControl;
+import org.opends.server.controls.PersistentSearchChangeType;
+import org.opends.server.controls.PersistentSearchControl;
import org.opends.server.core.BackendConfigManager;
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.ModifyDNOperation;
import org.opends.server.core.ModifyDNOperationBasis;
import org.opends.server.core.ModifyOperation;
+import org.opends.server.core.PersistentSearch;
import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.protocols.internal.InternalSearchListener;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.Requests;
import org.opends.server.protocols.internal.SearchRequest;
@@ -103,6 +109,7 @@
import org.opends.server.types.Operation;
import org.opends.server.types.SearchFilter;
import org.opends.server.types.SearchResultEntry;
+import org.opends.server.types.SearchResultReference;
import org.opends.server.util.LDIFWriter;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
@@ -356,6 +363,104 @@
}
/**
+ * A change the changelog already holds can reach its replication server again: a directory
+ * server publishes it again, or a peer replication server sends it from its own changelog. The
+ * replica DB does not take it, and a cookie based persistent search must not be notified of it a
+ * second time.
+ */
+ @Test
+ public void aChangeTheChangelogAlreadyHoldsIsNotNotifiedAgainToACookieBasedPersistentSearch() throws Exception
+ {
+ String test = "CookiePersistentSearchNotifiedOnce";
+ debugInfo(test, "Starting test");
+
+ final List<CSN> notified = new CopyOnWriteArrayList<>();
+ final InternalSearchOperation searchOp =
+ startChangesOnlyCookieBasedPersistentSearch("(targetdn=*" + test + "*,o=test)", notified);
+ try
+ {
+ final CSN[] csns = generateCSNs(2, server1);
+ final UpdateMsg sentAgain = generateDeleteMsg(server1, csns[0], test, 1);
+ publishUpdateMessagesInOTest(test, false, sentAgain, sentAgain, generateDeleteMsg(server1, csns[1], test, 2));
+
+ // the replication server handles the changes of a session in order: once the last one is
+ // notified, the one sent again has been handled
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertThat(notified).contains(csns[1]);
+ }
+ });
+ assertThat(notified).as("the changes the persistent search was notified of")
+ .containsExactly(csns[0], csns[1]);
+ }
+ finally
+ {
+ cancelPersistentSearch(searchOp);
+ }
+ debugInfo(test, "Ending test successfully");
+ }
+
+ /**
+ * Starts a persistent search of the changes only, in cookie mode, which records the CSN of every
+ * change it is notified of.
+ */
+ private InternalSearchOperation startChangesOnlyCookieBasedPersistentSearch(String filterString,
+ final List<CSN> notified) throws Exception
+ {
+ final SearchRequest request = newSearchRequest(filterString)
+ .addControl(createCookieControl(""))
+ .addControl(new PersistentSearchControl(EnumSet.allOf(PersistentSearchChangeType.class), true, false));
+ final InternalSearchOperation searchOp = connection.processSearch(request, new InternalSearchListener()
+ {
+ @Override
+ public void handleInternalSearchEntry(InternalSearchOperation searchOperation, SearchResultEntry searchEntry)
+ {
+ notified.add(new CSN(getAttributeValue(searchEntry, "replicationcsn")));
+ }
+
+ @Override
+ public void handleInternalSearchReference(InternalSearchOperation searchOperation,
+ SearchResultReference searchReference)
+ {
+ // the changelog holds no reference
+ }
+ });
+ assertThat(findPersistentSearch(searchOp)).as(searchOp.getErrorMessage().toString()).isNotNull();
+ return searchOp;
+ }
+
+ private void cancelPersistentSearch(InternalSearchOperation searchOp)
+ {
+ final PersistentSearch pSearch = findPersistentSearch(searchOp);
+ if (pSearch != null)
+ {
+ pSearch.cancel();
+ }
+ }
+
+ private PersistentSearch findPersistentSearch(InternalSearchOperation searchOp)
+ {
+ for (PersistentSearch pSearch : ChangelogBackend.getInstance().getPersistentSearches())
+ {
+ // the persistent search holds the operation the backend processed, which wraps the given one
+ final Operation op = pSearch.getSearchOperation();
+ if (op.getClientConnection() == searchOp.getClientConnection()
+ && op.getOperationID() == searchOp.getOperationID())
+ {
+ return pSearch;
+ }
+ }
+ return null;
+ }
+
+ /**
* This test enables a second suffix. It will break all tests using search on
* one suffix if run before them, so it is necessary to add them as
* dependencies.
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java
new file mode 100644
index 0000000..96c866e
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java
@@ -0,0 +1,868 @@
+/*
+ * 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 org.assertj.core.api.Assertions.assertThat;
+import static org.forgerock.opendj.ldap.ModificationType.REPLACE;
+import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
+import static org.opends.server.util.CollectionUtils.newArrayList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeSet;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.core.DirectoryServer;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.replication.protocol.ReplicaOfflineMsg;
+import org.opends.server.replication.protocol.UpdateMsg;
+import org.opends.server.replication.server.changelog.api.ChangelogException;
+import org.opends.server.replication.server.changelog.api.DBCursor;
+import org.opends.server.types.Attributes;
+import org.opends.server.types.Modification;
+import org.opends.server.util.TimeThread;
+import org.testng.annotations.Test;
+
+/**
+ * A handler which is behind reads the changes for its consumer from the changelog, and goes back
+ * to its in-memory queue once the queue holds what the consumer lacks - see issue #1206. The queue
+ * is ordered by CSN and drops its oldest changes when it is full, so holding the oldest change the
+ * consumer lacks does not mean holding all of them. A change which reaches the queue late with an
+ * older CSN - relayed by another replication server, or published again by a recovery - can be
+ * the oldest change the consumer lacks while the queue lacks newer changes of another replica: the
+ * ones it dropped before that change came, or the ones which reached the domain before the handler
+ * did. Served from the queue then, the consumer would be given the changes of that replica which
+ * the queue still holds, and its state would move past the ones in between for good.
+ * <p>
+ * Each test drives a handler which no domain feeds, from a single thread, as the writer of a
+ * session does. A change "arrives" the way {@code ReplicationServerDomain.put()} hands it over to
+ * a registered handler: written to the changelog, then given to the queue. A change written to the
+ * changelog only stands for one which reached the domain before the handler was registered.
+ * Replica {@link #A} stands for a directory server connected to the replication server, replica
+ * {@link #B} for one whose changes another replication server relays late: the changes of B carry
+ * older CSNs than the changes of A which arrive with them.
+ */
+@SuppressWarnings("javadoc")
+public class MessageHandlerCatchUpTest extends ReplicationTestCase
+{
+ private static final int RS_ID = 120;
+ private static final int A = 121;
+ private static final int B = 122;
+ private static final int C = 123;
+ private static final int WINDOW_SIZE = 100;
+ /** A queue no test fills. */
+ private static final int LARGE_QUEUE_SIZE = 100;
+ /** How long the handler is given to come up with a change it has: it reads them from memory. */
+ private static final long NEXT_MESSAGE_TIMEOUT_MS = 5000;
+
+ /**
+ * The case of the description of #1206, on a handler connected for long: the queue, full, has
+ * dropped a change of A the consumer lacks, and a change of B with an older CSN has come in
+ * after, into the room the drop left. The change of B is the oldest change the consumer lacks,
+ * and it is in the queue - but the change of A is not, and must still be sent.
+ */
+ @Test
+ public void aChangeTheFullQueueDroppedIsSentWhenAnOlderOneCameInAfterIt() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerDroppedDb");
+ final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, A, true);
+ final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, queue.consumerState);
+
+ queue.arriveIn(replicationServer, handler);
+
+ assertThat(csnsOf(nextMessages(handler, queue.changesTheConsumerLacks.size())))
+ .as("the consumer was not sent every change it lacks, in order")
+ .containsExactlyElementsOf(csnsOf(queue.changesTheConsumerLacks));
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The case of the reconnections of #1206: a new handler, whose queue does not hold the changes
+ * of A which reached the domain before it did. A change of B with an older CSN then arrives with
+ * a change of A: the change of B is the oldest change the consumer lacks and it is in the queue,
+ * but the earlier changes of A are not, and must still be sent.
+ */
+ @Test
+ public void aChangeWhichReachedTheDomainBeforeTheHandlerIsSentWhenAnOlderOneIsQueued() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerBeforeDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
+ final List<UpdateMsg> earlierChangesOfA = changesOfA(baseDN, t, 1, 3);
+ final DeleteMsg a4 = newDeleteMsg(baseDN, "a4", new CSN(t + 24, 4, A));
+ // the consumer has the changes of B up to b0, and none of A
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0));
+
+ publishToChangelogOnly(replicationServer, baseDN, b0);
+ publishToChangelogOnly(replicationServer, baseDN, earlierChangesOfA);
+ arrive(replicationServer, handler, a4);
+ arrive(replicationServer, handler, b1);
+
+ final List<UpdateMsg> expected = new ArrayList<>();
+ expected.add(b1);
+ expected.addAll(earlierChangesOfA);
+ expected.add(a4);
+ assertThat(csnsOf(nextMessages(handler, expected.size())))
+ .as("the consumer was not sent every change it lacks, in order")
+ .containsExactlyElementsOf(csnsOf(expected));
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The same as {@link #aChangeWhichReachedTheDomainBeforeTheHandlerIsSentWhenAnOlderOneIsQueued()}
+ * for a replica which writes nothing more once the handler is there: the queue is handed changes
+ * of B only, never one of A, and the earlier changes of A must still be sent.
+ */
+ @Test
+ public void aChangeOfAReplicaTheQueueIsNeverHandedIsSentWhenAnOlderOneIsQueued() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerQuietDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
+ final List<UpdateMsg> earlierChangesOfA = changesOfA(baseDN, t, 1, 3);
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0));
+
+ publishToChangelogOnly(replicationServer, baseDN, earlierChangesOfA);
+ arrive(replicationServer, handler, b0);
+ arrive(replicationServer, handler, b1);
+
+ final List<UpdateMsg> expected = new ArrayList<>();
+ expected.add(b1);
+ expected.addAll(earlierChangesOfA);
+ assertThat(csnsOf(nextMessages(handler, expected.size())))
+ .as("the consumer was not sent every change it lacks, in order")
+ .containsExactlyElementsOf(csnsOf(expected));
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A change of B with an older CSN arrives while the handler reads the changelog, too late for the
+ * read: the first change read is a change of A, and it is in the queue - but the change of B
+ * before it is not one the consumer has, and must still be sent.
+ */
+ @Test
+ public void aChangeWhichArrivedWhileTheChangelogWasReadIsSent() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ ChangeAfterTheRead domain = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerWhileReadDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final DeleteMsg a0 = newDeleteMsg(baseDN, "a0", new CSN(t + 10, 0, A));
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
+ final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 2);
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0, a0));
+ domain = new ChangeAfterTheRead(replicationServer, handler, b1);
+
+ arrive(replicationServer, handler, b0);
+ arrive(replicationServer, handler, a0);
+ for (UpdateMsg msg : changesOfA)
+ {
+ arrive(replicationServer, handler, msg);
+ }
+
+ final List<UpdateMsg> expected = new ArrayList<>(changesOfA);
+ expected.add(b1);
+ assertThat(csnsOf(nextMessages(handler, expected.size())))
+ .as("the consumer was not sent every change it lacks")
+ .containsExactlyElementsOf(csnsOf(expected));
+ }
+ finally
+ {
+ deregister(domain);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The same as {@link #aChangeWhichArrivedWhileTheChangelogWasReadIsSent()} for a handler whose
+ * consumer has everything the changelog held: the read gives nothing, and the change of B, which
+ * arrives after it, is the oldest change of the full queue, which drops it at once. The queue
+ * holds nothing the consumer lacks, but the change of B must still be sent.
+ */
+ @Test
+ public void aChangeTheFullQueueDroppedWhileTheChangelogWasReadIsSent() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ ChangeAfterTheRead domain = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerDroppedWhileReadDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
+ final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 5);
+ final ServerState consumerState = stateOf(b0, changesOfA.get(changesOfA.size() - 1));
+ // full with the changes of A: one more change makes it drop its oldest one
+ final MessageHandler handler = newHandler(replicationServer, baseDN, changesOfA.size(), consumerState);
+ domain = new ChangeAfterTheRead(replicationServer, handler, b1);
+
+ for (UpdateMsg msg : changesOfA)
+ {
+ arrive(replicationServer, handler, msg);
+ }
+
+ assertThat(csnsOf(nextMessages(handler, 1)))
+ .as("the consumer was not sent the change the queue dropped")
+ .containsExactly(b1.getCSN());
+ }
+ finally
+ {
+ deregister(domain);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The same queue as {@link #aChangeTheFullQueueDroppedIsSentWhenAnOlderOneCameInAfterIt()}, for
+ * a consumer which already has the change of A the queue dropped: the queue holds everything the
+ * consumer lacks, and the handler serves it from the queue again, as soon as it finds there the
+ * oldest change the consumer lacks.
+ */
+ @Test
+ public void aHandlerWhoseQueueHoldsWhatTheConsumerLacksFollowsTheQueueAgain() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerFollowsDb");
+ final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, A, true);
+ final ServerState consumerState = queue.consumerState.duplicate();
+ consumerState.update(queue.dropped.getCSN());
+ final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, consumerState);
+
+ queue.arriveIn(replicationServer, handler);
+
+ final List<UpdateMsg> expected = new ArrayList<>(queue.changesTheConsumerLacks);
+ expected.remove(queue.dropped);
+ assertServedFromTheQueueAfterTheFirst(handler, expected);
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The queue dropped a change the changelog no longer holds, as after the changelog was cleared
+ * for a new generation id, of a replica which writes nothing more: nothing can send it to the
+ * consumer, and it must not keep the handler from its queue, which holds everything else the
+ * consumer lacks.
+ */
+ @Test
+ public void aDroppedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerNoLongerHeldDb");
+ final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, C, false);
+ final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, queue.consumerState);
+
+ queue.arriveIn(replicationServer, handler);
+
+ final List<UpdateMsg> expected = new ArrayList<>(queue.changesTheConsumerLacks);
+ expected.remove(queue.dropped);
+ assertServedFromTheQueueAfterTheFirst(handler, expected);
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A replica offline message carries the CSN of no change, and no state of a consumer ever holds
+ * it: one in the queue before the first change the consumer lacks must not keep the handler from
+ * its queue, even once its replica, back online, has written a change the consumer lacks.
+ */
+ @Test
+ public void aReplicaOfflineMessageDoesNotHoldTheHandlerBack() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerOfflineDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg c0 = newDeleteMsg(baseDN, "c0", new CSN(t, 0, C));
+ final DeleteMsg a1 = newDeleteMsg(baseDN, "a1", new CSN(t + 1, 1, A));
+ final DeleteMsg a2 = newDeleteMsg(baseDN, "a2", new CSN(t + 3, 2, A));
+ final DeleteMsg c3 = newDeleteMsg(baseDN, "c3", new CSN(t + 4, 3, C));
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(c0, a1));
+
+ arrive(replicationServer, handler, c0);
+ arrive(replicationServer, handler, a1);
+ // a replica offline message is not written to the changelog, its CSN goes to the state of
+ // the replica instead: ReplicationServerDomain.publishUpdateMsg()
+ handler.add(new ReplicaOfflineMsg(new CSN(t + 2, 0, C)));
+ arrive(replicationServer, handler, a2);
+ arrive(replicationServer, handler, c3);
+
+ final List<UpdateMsg> expected = new ArrayList<>();
+ expected.add(a2);
+ expected.add(c3);
+ assertServedFromTheQueueAfterTheFirst(handler, expected);
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The queue still holds a change the changelog no longer holds, as after the changelog was
+ * cleared for a new generation id, and the changelog gives the handler nothing but the replica
+ * offline message of a replica which wrote since: no read of the changelog can give the consumer
+ * that change, and the handler must not read the changelog again and again for it. It goes back
+ * to its queue, leaving out what the queue holds before the offline message, that change too.
+ */
+ @Test
+ public void aQueuedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerQueuedNoLongerHeldDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg c0 = newDeleteMsg(baseDN, "c0", new CSN(t, 0, C));
+ final DeleteMsg a1 = newDeleteMsg(baseDN, "a1", new CSN(t + 1, 1, A));
+ final CSN offlineC = new CSN(t + 2, 0, C);
+ final DeleteMsg a2 = newDeleteMsg(baseDN, "a2", new CSN(t + 3, 2, A));
+ // the consumer has every change the changelog holds
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(c0));
+
+ publishToChangelogOnly(replicationServer, baseDN, c0);
+ // the change of a changelog cleared since
+ handler.add(a1);
+ // every read of the changelog gives it, as no state of a consumer ever holds its CSN
+ replicationServer.getChangelogDB().getReplicationDomainDB().notifyReplicaOffline(baseDN, offlineC);
+
+ assertThat(csnsOf(nextMessages(handler, 1)))
+ .as("the handler kept reading the changelog for a change no read can give")
+ .containsExactly(offlineC);
+ assertThat(handler.isFollowing()).isTrue();
+
+ arrive(replicationServer, handler, a2);
+ final List<Sent> sent = nextMessages(handler, 1);
+ assertThat(csnsOf(sent)).containsExactly(a2.getCSN());
+ assertThat(sent.get(0).fromTheChangelog).isFalse();
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The full queue of a handler drops a replica offline message, which the domain hands to the
+ * handler of a peer replication server: the consumer has every change of that replica the
+ * changelog holds, and no state of a consumer ever holds the CSN of the message, so it must not
+ * keep the handler from its queue, which holds every change the consumer lacks.
+ */
+ @Test
+ public void aDroppedReplicaOfflineMessageDoesNotHoldTheHandlerBack() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerDroppedOfflineDb");
+ final long t = TimeThread.getTime() - 60000;
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 5);
+ // full with the changes of A: they make it drop the offline message, its oldest one
+ final MessageHandler handler = newHandler(replicationServer, baseDN, changesOfA.size(), stateOf(b0));
+
+ publishToChangelogOnly(replicationServer, baseDN, b0);
+ handler.add(new ReplicaOfflineMsg(new CSN(t + 5, 0, B)));
+ for (UpdateMsg msg : changesOfA)
+ {
+ arrive(replicationServer, handler, msg);
+ }
+
+ assertServedFromTheQueueAfterTheFirst(handler, changesOfA);
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A handler which has been handed a change catches up from the changelog until a read gives
+ * nothing, and goes back to its queue: the change handed to it after that read is served from
+ * the queue.
+ */
+ @Test
+ public void aHandlerWhichCaughtUpFromTheChangelogFollowsTheQueueAgain() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ ChangeAfterTheRead domain = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerCaughtUpDb");
+ final long t = TimeThread.getTime() - 60000;
+ final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 3);
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, new ServerState());
+ domain = new ChangeAfterTheRead(replicationServer, handler, changesOfA.get(2), true);
+
+ publishToChangelogOnly(replicationServer, baseDN, changesOfA.get(0));
+ arrive(replicationServer, handler, changesOfA.get(1));
+
+ assertServedFromTheQueueAfterAReadGaveNothing(handler, changesOfA);
+ }
+ finally
+ {
+ deregister(domain);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The same as {@link #aHandlerWhichCaughtUpFromTheChangelogFollowsTheQueueAgain()} for a handler
+ * which is never handed the changes of a replica, as the handler of a peer replication server is
+ * never handed the changes of the replicas connected to other replication servers: the domain held
+ * a change of B when it handed the queue its first change, so the handler reads it from the
+ * changelog, although it finds in its queue the first change it reads, then goes back to its queue.
+ */
+ @Test
+ public void aHandlerWhichReadTheChangesOfAReplicaItIsNeverHandedFollowsTheQueueAgain() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ ChangeAfterTheRead domain = null;
+ try
+ {
+ replicationServer = newReplicationServer("messageHandlerNeverHandedDb");
+ final long t = TimeThread.getTime() - 60000;
+ final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 2);
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 30, 1, B));
+ final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, new ServerState());
+ domain = new ChangeAfterTheRead(replicationServer, handler, changesOfA.get(1), true);
+
+ publishToChangelogOnly(replicationServer, baseDN, b1);
+ arrive(replicationServer, handler, changesOfA.get(0));
+
+ final List<UpdateMsg> expected = new ArrayList<>();
+ expected.add(changesOfA.get(0));
+ expected.add(b1);
+ expected.add(changesOfA.get(1));
+ assertServedFromTheQueueAfterAReadGaveNothing(handler, expected);
+ }
+ finally
+ {
+ deregister(domain);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A queue of {@link #size} changes which drops a change the consumer lacks, then takes a change
+ * of B with an older CSN in the room the drop left - which only a queue bounded by its size in
+ * bytes leaves, as it drops whole changes: one large change, dropped, leaves room for a small one.
+ * <p>
+ * The queue is not full until it holds 5 changes. Changes {@code b0} and {@code a0}, which the
+ * consumer has, show that the handler has been handed changes for long; the large change pushes
+ * the queue over its size with the fifth change, so the queue drops {@code b0}, then {@code a0},
+ * then the large change. The changes after it are small, so {@code b1} comes into the room the
+ * large change left: the queue ends up holding {@code b1} and {@code a2} .. {@code a6}.
+ */
+ private final class DroppingQueue
+ {
+ private final List<UpdateMsg> arrivals = new ArrayList<>();
+ /** The changes the consumer lacks, in the order it must be sent them. */
+ private final List<UpdateMsg> changesTheConsumerLacks = new ArrayList<>();
+ private final ServerState consumerState;
+ private final UpdateMsg dropped;
+ private final boolean droppedInTheChangelog;
+ private final int size;
+
+ /**
+ * @param droppedReplica
+ * the replica of the change the queue drops
+ * @param droppedInTheChangelog
+ * whether the changelog holds the change the queue drops
+ */
+ private DroppingQueue(DN baseDN, long t, int droppedReplica, boolean droppedInTheChangelog)
+ throws Exception
+ {
+ final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
+ final DeleteMsg a0 = newDeleteMsg(baseDN, "a0", new CSN(t + 10, 0, A));
+ final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
+ // a replace of a large value: the size of a delete does not count its DN
+ dropped = new ModifyMsg(new CSN(t + 21, 1, droppedReplica), DN.valueOf("cn=large," + baseDN),
+ newArrayList(new Modification(REPLACE, Attributes.create("description", "a1".repeat(1000)))),
+ "uuid");
+ this.droppedInTheChangelog = droppedInTheChangelog;
+ final List<UpdateMsg> keptChangesOfA = changesOfA(baseDN, t, 2, 6);
+
+ arrivals.add(b0);
+ arrivals.add(a0);
+ arrivals.add(dropped);
+ arrivals.addAll(keptChangesOfA);
+ arrivals.add(b1);
+ changesTheConsumerLacks.add(b1);
+ changesTheConsumerLacks.add(dropped);
+ changesTheConsumerLacks.addAll(keptChangesOfA);
+ consumerState = stateOf(b0, a0);
+
+ // room for b1 and a2 .. a6 in bytes, as many changes as the queue ever holds, and not for the
+ // large change
+ int keptBytes = b1.size();
+ for (UpdateMsg msg : keptChangesOfA)
+ {
+ keptBytes += msg.size();
+ }
+ size = Math.max(arrivals.size(), (keptBytes + 99) / 100);
+ assertThat(dropped.size())
+ .as("the large change must push the queue over its size in bytes on its own")
+ .isGreaterThan(size * 100);
+ }
+
+ private void arriveIn(ReplicationServer replicationServer, MessageHandler handler) throws Exception
+ {
+ for (UpdateMsg msg : arrivals)
+ {
+ if (msg != dropped || droppedInTheChangelog)
+ {
+ arrive(replicationServer, handler, msg);
+ }
+ else
+ {
+ handler.add(msg);
+ }
+ }
+ }
+ }
+
+ /**
+ * The domain of a handler, which hands it a change once it has read the changelog for the first
+ * time - or once a read of the changelog gave nothing - before it looks at its queue: the domain
+ * writes the changes it receives and hands them over on threads of their own, and nothing keeps
+ * one from coming in between. The change comes the way {@code ReplicationServerDomain.put()}
+ * hands it over - see {@link #arrive}.
+ */
+ private final class ChangeAfterTheRead extends ReplicationServerDomain
+ {
+ private final ReplicationServer replicationServer;
+ private final MessageHandler handler;
+ private final boolean onceAReadGivesNothing;
+ private UpdateMsg change;
+
+ private ChangeAfterTheRead(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg change)
+ {
+ this(replicationServer, handler, change, false);
+ }
+
+ /**
+ * @param onceAReadGivesNothing
+ * whether the change comes once a read of the changelog gave nothing, rather than once
+ * the changelog was read for the first time
+ */
+ private ChangeAfterTheRead(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg change,
+ boolean onceAReadGivesNothing)
+ {
+ super(handler.getBaseDN(), replicationServer);
+ this.replicationServer = replicationServer;
+ this.handler = handler;
+ this.change = change;
+ this.onceAReadGivesNothing = onceAReadGivesNothing;
+ handler.replicationServerDomain = this;
+ }
+
+ @Override
+ public String getMonitorInstanceName()
+ {
+ // apart from the one of the domain of the replication server, which it would replace
+ return super.getMonitorInstanceName() + " for " + MessageHandlerCatchUpTest.class.getSimpleName();
+ }
+
+ @Override
+ public DBCursor<UpdateMsg> getCursorFrom(ServerState startAfterServerState) throws ChangelogException
+ {
+ final DBCursor<UpdateMsg> read = super.getCursorFrom(startAfterServerState);
+ return new DBCursor<UpdateMsg>()
+ {
+ private boolean gaveSomething;
+
+ @Override
+ public UpdateMsg getRecord()
+ {
+ return read.getRecord();
+ }
+
+ @Override
+ public boolean next() throws ChangelogException
+ {
+ final boolean hasNext = read.next();
+ gaveSomething |= hasNext;
+ return hasNext;
+ }
+
+ @Override
+ public void close()
+ {
+ read.close();
+ if (change != null && (!onceAReadGivesNothing || !gaveSomething))
+ {
+ final UpdateMsg msg = change;
+ change = null;
+ try
+ {
+ arrive(replicationServer, handler, msg);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ };
+ }
+ }
+
+ private static void deregister(ReplicationServerDomain domain)
+ {
+ if (domain != null)
+ {
+ DirectoryServer.deregisterMonitorProvider(domain);
+ }
+ }
+
+ /** A message the handler came up with, and whether it read it from the changelog. */
+ private static final class Sent
+ {
+ private final UpdateMsg msg;
+ private final boolean fromTheChangelog;
+
+ private Sent(UpdateMsg msg, boolean fromTheChangelog)
+ {
+ this.msg = msg;
+ this.fromTheChangelog = fromTheChangelog;
+ }
+ }
+
+ /**
+ * The handler sends the given changes, in order, and serves them from its queue from the second
+ * one on: it finds the first one in its queue and goes back to it.
+ */
+ private void assertServedFromTheQueueAfterTheFirst(MessageHandler handler, List<UpdateMsg> expected)
+ throws Exception
+ {
+ final List<Sent> sent = nextMessages(handler, expected.size());
+ assertThat(csnsOf(sent)).containsExactlyElementsOf(csnsOf(expected));
+ assertThat(sent.get(1).fromTheChangelog)
+ .as("the handler did not go back to its queue, which holds every change the consumer lacks")
+ .isFalse();
+ assertThat(handler.isFollowing()).isTrue();
+ }
+
+ /**
+ * The handler sends the given changes, in order, and serves the last one from its queue: it is
+ * handed to the queue once a read of the changelog gave nothing, see {@link ChangeAfterTheRead}.
+ */
+ private void assertServedFromTheQueueAfterAReadGaveNothing(MessageHandler handler, List<UpdateMsg> expected)
+ throws Exception
+ {
+ final List<Sent> sent = nextMessages(handler, expected.size());
+ assertThat(csnsOf(sent)).containsExactlyElementsOf(csnsOf(expected));
+ assertThat(sent.get(sent.size() - 1).fromTheChangelog)
+ .as("the handler did not go back to its queue once the changelog had nothing more for the consumer")
+ .isFalse();
+ assertThat(handler.isFollowing()).isTrue();
+ }
+
+ /**
+ * The next messages the handler comes up with for its consumer, at most {@code count}: fewer
+ * when it has none to come up with. It is asked from one thread, as the writer of a session asks.
+ */
+ private List<Sent> nextMessages(final MessageHandler handler, int count) throws Exception
+ {
+ final ExecutorService writer = Executors.newSingleThreadExecutor();
+ try
+ {
+ final List<Sent> sent = new ArrayList<>();
+ for (int i = 0; i < count; i++)
+ {
+ final Future<Sent> next = writer.submit(new Callable<Sent>()
+ {
+ @Override
+ public Sent call() throws Exception
+ {
+ final UpdateMsg msg = handler.getNextMessage();
+ return msg != null ? new Sent(msg, handler.isLastMessageFromLateQueue()) : null;
+ }
+ });
+ final Sent nextSent;
+ try
+ {
+ nextSent = next.get(NEXT_MESSAGE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ // it waits for a change to reach its queue: it has nothing more for the consumer
+ handler.setConsumerActive(false);
+ break;
+ }
+ if (nextSent == null)
+ {
+ break;
+ }
+ sent.add(nextSent);
+ }
+ return sent;
+ }
+ finally
+ {
+ writer.shutdownNow();
+ }
+ }
+
+ private MessageHandler newHandler(ReplicationServer replicationServer, DN baseDN, int queueSize,
+ ServerState consumerState) throws Exception
+ {
+ final MessageHandler handler = new MessageHandler(queueSize, replicationServer);
+ handler.setBaseDNAndDomain(baseDN, false);
+ handler.setInitialServerState(consumerState);
+ return handler;
+ }
+
+ /** Hands a change over the way {@code ReplicationServerDomain.put()} does. */
+ private void arrive(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg msg)
+ throws Exception
+ {
+ publishToChangelogOnly(replicationServer, handler.getBaseDN(), msg);
+ handler.add(msg);
+ }
+
+ private void publishToChangelogOnly(ReplicationServer replicationServer, DN baseDN, List<UpdateMsg> msgs)
+ throws Exception
+ {
+ for (UpdateMsg msg : msgs)
+ {
+ publishToChangelogOnly(replicationServer, baseDN, msg);
+ }
+ }
+
+ private void publishToChangelogOnly(ReplicationServer replicationServer, DN baseDN, UpdateMsg msg)
+ throws Exception
+ {
+ replicationServer.getChangelogDB().getReplicationDomainDB().publishUpdateMsg(baseDN, msg);
+ }
+
+ /** Changes {@code a<first>} .. {@code a<last>} of A, a second apart, the first 20 s after B's first one. */
+ private List<UpdateMsg> changesOfA(DN baseDN, long t, int first, int last) throws Exception
+ {
+ final List<UpdateMsg> changes = new ArrayList<>();
+ for (int i = first; i <= last; i++)
+ {
+ changes.add(newDeleteMsg(baseDN, "a" + i, new CSN(t + 20 + i, i, A)));
+ }
+ return changes;
+ }
+
+ private static ServerState stateOf(UpdateMsg... msgs)
+ {
+ final ServerState state = new ServerState();
+ for (UpdateMsg msg : msgs)
+ {
+ state.update(msg.getCSN());
+ }
+ return state;
+ }
+
+ private static List<CSN> csnsOf(List<?> msgs)
+ {
+ final List<CSN> csns = new ArrayList<>();
+ for (Object msg : msgs)
+ {
+ csns.add(msg instanceof Sent ? ((Sent) msg).msg.getCSN() : ((UpdateMsg) msg).getCSN());
+ }
+ return csns;
+ }
+
+ private DeleteMsg newDeleteMsg(DN baseDN, String rdnValue, CSN csn) throws Exception
+ {
+ return new DeleteMsg(DN.valueOf("cn=" + rdnValue + "," + baseDN), csn, "uuid");
+ }
+
+ private ReplicationServer newReplicationServer(String dbDirName) throws Exception
+ {
+ return new ReplicationServer(new ReplServerFakeConfiguration(
+ TestCaseUtils.findFreePort(), dbDirName, 0, RS_ID, 0, WINDOW_SIZE, new TreeSet<String>()));
+ }
+
+ private void removeQuietly(ReplicationServer replicationServer)
+ {
+ try
+ {
+ if (replicationServer != null)
+ {
+ remove(replicationServer);
+ }
+ }
+ catch (Exception ignored)
+ {
+ // the test has already reported what matters
+ }
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
index 5c084ca..aee0ad1 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java
@@ -12,6 +12,7 @@
* information: "Portions Copyright [year] [name of copyright owner]".
*
* Copyright 2014-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server.changelog.file;
@@ -82,6 +83,29 @@
return Log.openLog(replicationEnv, LOG_DIRECTORY, parser, rotationParams);
}
+ /**
+ * A record whose key is not after the newest one is not appended, whether the head log file is
+ * to be rotated or not, and the log tells so.
+ */
+ @Test
+ public void appendTellsWhetherTheRecordWasAppended() throws Exception
+ {
+ try (Log<String, String> log = openLog(RECORD_PARSER))
+ {
+ // the head log file holds 2 records: the next append rotates it
+ assertThat(log.append(Record.from("key010", "value10 again"))).as("the newest key again").isFalse();
+ assertThat(log.append(Record.from("key005", "value5 again"))).as("an older key").isFalse();
+ assertThat(log.append(Record.from("key011", "value11"))).as("a newer key").isTrue();
+ // the new head log file holds 1 record: no rotation
+ assertThat(log.append(Record.from("key011", "value11 again"))).as("the newest key again").isFalse();
+
+ try (DBCursor<Record<String, String>> cursor = log.getCursor())
+ {
+ assertThatCursorCanBeFullyReadFromStart(cursor, 1, 11);
+ }
+ }
+ }
+
@Test
public void testCursor() throws Exception
{
--
Gitblit v1.10.0