[#1206] Take a late handler of a replication server back to its queue only ...

Valery Kharseko

2 days ago b8a4d35f03b7fa5cf0d80a9424224f6ddeb3b3f9

 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;
              }

 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()
  {

 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)

 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;
  }

  /**

 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
    {

 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
    {

 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.

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java

New file
@@ -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
    }
  }
}

 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
  {