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/backends/ChangelogBackendTestCase.java | 105 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 105 insertions(+), 0 deletions(-)
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.
--
Gitblit v1.10.0