Send what a replication session's publisher left queued when it is closed (...

Valery Kharseko

13 hours ago bf0f4f294f477e393b59ef7f160795e00aa02cc9

Send what a replication session's publisher left queued when it is closed (#1035)

 opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2006-2009 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.protocol;

@@ -30,12 +31,14 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.zip.DataFormatException;

import javax.net.ssl.SSLSocket;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.api.DirectoryThread;
import org.opends.server.types.HostPort;
@@ -48,6 +51,19 @@
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /**
   * How long a close spends sending what the publisher thread left queued in {@code sendQueue},
   * in milliseconds.
   * <p>
   * The same 5 s as {@code DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD}, which is how long a
   * shutdown is already willing to wait for one of these messages - the announcement that a
   * replica went offline - to be forwarded. A close has no reason to wait longer for it than the
   * shutdown which is waiting on the close. It is a budget per close, though, not per shutdown:
   * a shutdown closes its sessions one after another, so it can pay this once for each peer which
   * is alive and not reading.
   */
  private static final long DRAIN_BUDGET_MS = 5000;

  private final Socket plainSocket;
  private final SSLSocket secureSocket;
  private final InputStream plainInput;
@@ -101,6 +117,14 @@
  private final CountDownLatch latch = new CountDownLatch(1);

  /**
   * How many buffers the publisher thread took off {@code sendQueue} and failed to write. After a
   * failed write that thread goes on taking the queue until the session is closed, and every
   * write after the first fails as well, so these are messages the peer was not told about just
   * as much as what is still queued - see {@link #close()}.
   */
  private final AtomicInteger publisherFailedWrites = new AtomicInteger();

  /**
   * Creates a new Session.
   *
   * @param socket
@@ -140,6 +164,10 @@
  /**
   * This method is called when the session with the remote must be closed.
   * This object won't be used anymore after this method is called.
   * <p>
   * A message which was published on this session but which its publisher thread had not sent yet
   * is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}. See {@link
   * #sendWhatThePublisherLeftQueued()}.
   */
  @Override
  public void close()
@@ -165,6 +193,18 @@
      Thread.currentThread().interrupt();
    }

    /*
     * Re-read the error rather than answer with the snapshot taken before the join: a publisher
     * whose send() failed while this thread was joining it recorded the error there, and what
     * follows - the drain and the StopMsg - is what must not be written to a socket which has
     * already failed. Reading it before the join left both writing to one, the drain naming its
     * own failure rather than the one the publisher had recorded.
     */
    synchronized (stateLock)
    {
      localSessionError = sessionError;
    }

    // Perform close outside of critical section.
    if (logger.isTraceEnabled())
    {
@@ -186,18 +226,72 @@
      }
    }

    // V4 protocol introduces a StopMsg to properly end communications.
    if (localSessionError == null
        && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
    if (localSessionError != null)
    {
      try
      /*
       * Nothing more is written to a session which already failed, neither what the publisher left
       * queued nor the StopMsg: writing more to it cannot work. What it gives up on is still
       * reported: what is left in the queue, and what the publisher took off it and failed to
       * write - the write which failed first, and every one it went on failing until this close
       * stopped it, which the join above makes complete. It is done without taking publishLock,
       * so that a thread blocked in a write of this socket is released by the close of the
       * sockets below rather than waited for.
       */
      isRunning.set(false);
      final int notSent = publisherFailedWrites.get() + takeWhatIsLeftQueued();
      if (notSent > 0)
      {
        publish(new StopMsg());
        reportQueueNotSent(notSent, "the session had already failed: " + localSessionError);
      }
      catch (final IOException ignored)
      StaticUtils.close(plainSocket, secureSocket);
      return;
    }

    /*
     * The publisher thread has stopped and what it had not sent is still in the queue. Send it,
     * rather than let the close drop it: nothing publishes these again, and the StopMsg which
     * follows leaves the peer reading an orderly close with no sign that anything was missing.
     *
     * This thread is not the only one which can write the socket here - a ServerWriter or a
     * HeartbeatThread outlives this close. It is this thread, not run(), which takes the session
     * off the queueing branch of publish(), and it does so under publishLock and keeps the lock
     * across the queue and the StopMsg: until then a publish() concurrent with the close takes
     * the queueing branch, and from then on it takes the synchronous branch and waits for the
     * lock. Either way nothing newer is written between two of the drained messages - see
     * sendWhatThePublisherLeftQueued() below.
     */
    publishLock.lock();
    try
    {
      isRunning.set(false);
      sendWhatThePublisherLeftQueued();

      /*
       * Re-read again: a write of the drain which failed recorded its error, and the StopMsg must
       * not be written to a socket which has already failed either.
       */
      synchronized (stateLock)
      {
        // Ignore errors on close.
        localSessionError = sessionError;
      }

      // V4 protocol introduces a StopMsg to properly end communications.
      if (localSessionError == null
          && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
      {
        try
        {
          publish(new StopMsg());
        }
        catch (final IOException ignored)
        {
          // Ignore errors on close.
        }
      }
    }
    finally
    {
      publishLock.unlock();
    }

    StaticUtils.close(plainSocket, secureSocket);
@@ -206,6 +300,101 @@


  /**
   * Sends the buffers the publisher thread had not sent when it stopped, so that a close of the
   * session does not drop them.
   * <p>
   * Called from {@link #close()} once the publisher has been joined, with {@code publishLock}
   * held. A queued message is already encoded for this peer's protocol version - {@link
   * #publish(ReplicationMsg)} did that before queueing it - so there is nothing to decide here
   * beyond how long to keep trying.
   * <p>
   * The whole queue goes out under {@code publishLock}, because the publisher is not the only
   * thread which writes this socket: a {@code ServerWriter} is joined only after the close which
   * gets here (ServerHandler.shutdown() closes the session before joining it, and ServerReader's
   * finally closes it before stopping the handler), and a {@code HeartbeatThread} is shut down
   * after it too. Their {@code publish()} takes the synchronous branch once the session is off
   * the queueing one, so without the lock a newer message could be written between two of these
   * older ones - which a peer replication server answers by dropping the older ones at debug
   * level, its log file refusing a record which would break its key ordering. The close takes
   * the session off the queueing branch under the same lock it holds here, so such a
   * {@code publish()} either queues ahead of that and is drained here, or waits for the drain and
   * lands after it, or returns at the door, having seen the close. One which read the close as
   * not yet begun can still be descheduled before its buffer is queued and queue it after the
   * last poll below; it checks for that once it has, and takes the buffer back and reports it -
   * see {@link #publish(ReplicationMsg)}. The wait for the lock itself is not part of the budget
   * below, no more than it is for the {@code StopMsg} which follows.
   * <p>
   * The budget bounds how many messages a close spends on a peer which is reading slowly. A peer
   * which is reading pays none of it; a peer which answers with a reset pays one failed write. It
   * is checked between messages, so a single write which blocks past the budget still runs to
   * completion - for a peer which has stopped reading, or which vanished without a reset, that is
   * for as long as TCP keeps the connection alive: bounding it needs a non-blocking socket, which
   * this session is not. The budget is per close, and a shutdown closes its sessions one after
   * another - ReplicationServerDomain.stopAllServers() stops each handler in turn on one thread -
   * so a domain with several peers which are alive and not reading pays it once per such peer.
   * On the road which does not shut the whole server down the wait is paid under the lock of the
   * replication domain - ReplicationServerDomain.stopServer() holds it across the handler
   * shutdown which closes this session - where a handshake meanwhile waiting on that lock times
   * out and the broker retries. What is given up on is reported rather than dropped in silence,
   * that being the part of this which cost the most to diagnose.
   */
  private void sendWhatThePublisherLeftQueued()
  {
    final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS);
    byte[] buffer;
    while ((buffer = sendQueue.poll()) != null)
    {
      if (System.nanoTime() - deadline >= 0)
      {
        reportQueueNotSent(1 + takeWhatIsLeftQueued(), "the peer did not read them within "
            + DRAIN_BUDGET_MS + " ms");
        return;
      }
      try
      {
        send(buffer);
      }
      catch (final IOException e)
      {
        /*
         * send() has recorded the error; the rest of the queue cannot go out either. The
         * exception is named rather than traced: this is reached whenever a peer which has
         * announced it is leaving closes before the drain reaches it, where what the write
         * failed with is the whole of what a reader of the log needs - a directory server
         * re-reads these from the changelog when it reconnects, a replication server does not.
         */
        reportQueueNotSent(1 + takeWhatIsLeftQueued(),
            "the write failed with " + e.getClass().getName() + ": " + e.getMessage());
        return;
      }
    }
  }

  /**
   * Empties the queue a close gives up on, and answers how many buffers it held. Taking them
   * rather than counting them is what keeps a {@code publish()} which queued a buffer late from
   * reporting it a second time: it reports only a buffer it still finds in the queue.
   */
  private int takeWhatIsLeftQueued()
  {
    int taken = 0;
    while (sendQueue.poll() != null)
    {
      taken++;
    }
    return taken;
  }

  /** Says which messages a close could not hand to the peer, and why. */
  private void reportQueueNotSent(final int count, final String reason)
  {
    logger.warn(LocalizableMessage.raw(
        "The replication session %s was closed with %d message(s) which had been published on it "
        + "but not yet sent, and the peer was not told about them: %s",
        getName(), count, reason));
  }

  /**
   * This methods allows to determine if the session close was initiated
   * on this Session.
   *
@@ -322,6 +511,10 @@
          // Avoid blocking forever so that we can check for session closure.
          if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS))
          {
            if (!isRunning.get())
            {
              takeBackWhatWasQueuedTooLate(buffer);
            }
            return;
          }
        }
@@ -338,6 +531,34 @@
    }
  }

  /**
   * Takes a buffer back out of the queue if nothing is going to send it, and reports it.
   * <p>
   * {@code publish()} reads the close as not yet begun and queues the buffer after that, with no
   * lock in between: descheduled there, it can queue the buffer after a close has stopped the
   * publisher thread and drained the queue, and nothing would then send it or say so. Once the
   * session is off the queueing branch, the lock here is the one the close holds while it takes
   * the session off that branch and drains the queue, so what is still in the queue after it is
   * what nothing sends. A buffer queued before the session came off the queueing branch is left
   * to the drain - it cannot have seen the flag cleared - and a buffer the drain or the close
   * already took is not found here, so nothing is reported twice.
   */
  private void takeBackWhatWasQueuedTooLate(final byte[] buffer)
  {
    publishLock.lock();
    try
    {
      if (sendQueue.remove(buffer))
      {
        reportQueueNotSent(1, "it was queued after the publisher of the session had stopped");
      }
    }
    finally
    {
      publishLock.unlock();
    }
  }

  /** Sends a replication message already encoded to the socket.
   *
   * @param buffer
@@ -526,7 +747,20 @@
  @Override
  public void run()
  {
    isRunning.set(true);
    synchronized (stateLock)
    {
      /*
       * A close which came before the start has already run, and it is not coming back to clear
       * the flag - nor is the end of this method, which leaves that to the close. Set, the flag
       * would keep every later publish() on the queueing branch, which returns at once on a
       * closed session as if the message had been queued; unset, publish() stays on the
       * synchronous branch, which fails on the closed socket.
       */
      if (!closeInitiated)
      {
        isRunning.set(true);
      }
    }
    latch.countDown();
    if (logger.isTraceEnabled())
    {
@@ -551,10 +785,21 @@
      catch (IOException e)
      {
        setSessionError(e);
        publisherFailedWrites.incrementAndGet();
        needClosing = true;
      }
    }
    isRunning.set(false);
    /*
     * A close clears the flag itself, under publishLock, once it has joined this thread - see
     * close(). Clearing it here would open a window between the end of this thread and the drain
     * of the close, in which a publish() takes the synchronous branch and writes a newer message
     * ahead of the queue. Only a loop which ended without a close - an interrupt from elsewhere -
     * clears it here, so that publish() does not go on queueing onto a queue nobody sends.
     */
    if (!closeInitiated)
    {
      isRunning.set(false);
    }
    if (needClosing)
    {
      close();

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

New file
@@ -0,0 +1,759 @@
/*
 * 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.protocol;

import static org.assertj.core.api.Assertions.assertThat;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;

import java.io.Closeable;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.StaticUtils;
import org.testng.annotations.Test;

/**
 * Which end of a replication session has a publisher thread, and what a close of the session
 * does to the messages that thread has not sent yet.
 * <p>
 * {@link Session#publish(ReplicationMsg)} has two branches, and which one a message takes used
 * to decide whether a close could lose it. With a publisher thread running the call is an enqueue
 * onto {@code sendQueue}; without one it is a synchronous write of the socket. {@link
 * Session#close()} used to drain neither: it set the flag the publisher loops on, interrupted it
 * and joined, so anything still queued was dropped while the {@code StopMsg} published afterwards
 * still went out - leaving the peer with an orderly close and no sign that something was lost.
 * That is the limitation PR #919 recorded, and
 * {@code aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed} is what holds the
 * close to sending that queue instead. The cases after it pin what a close gives up on: a
 * queue whose write fails is reported with every message it held; a session which had already
 * failed is written nothing more and still has its queue reported, together with what its
 * publisher took and failed to write, and nothing when there is nothing; and a closed session -
 * one closed before it was started included - fails a later {@code publish()} rather than take
 * it into a queue nothing sends.
 * <p>
 * The first two cases pin which end could ever pay it. Only {@code ServerHandler} starts a session's
 * publisher, so it is the replication-server end of a session which has one; the broker of a
 * directory server never starts its own. A change a directory server publishes is therefore on
 * the wire by the time {@code publish()} returns, and no close of that session could drop it -
 * which is what rules the send queue out as the explanation of #963.
 *
 * @see <a href="https://github.com/OpenIdentityPlatform/OpenDJ/issues/963">issue #963</a>
 */
@SuppressWarnings("javadoc")
public class SessionPublisherDrainTest extends ReplicationTestCase
{
  private static final int DS_ID = 123;
  private static final int RS_ID = 104;
  private static final int SOCKET_TIMEOUT_MS = 5000;

  /**
   * The number of messages the queued-message test publishes. It has to be enough for
   * {@code publish()} to outrun the publisher thread and leave a backlog behind it - the socket
   * buffers of a loopback pair swallow all of it, so it is not a full buffer which leaves one -
   * and stay under the 4000 the send queue holds, past which {@code publish()} would block
   * instead of queueing.
   */
  private static final int MESSAGES_PUBLISHED = 3000;

  /**
   * The number of messages the case of a queue which cannot be sent leaves on the sender. It is
   * small and exact: what that case is about is the count the close reports, not a backlog.
   */
  private static final int MESSAGES_LEFT_UNSENT = 7;

  /** What the line a close writes about a queue it could not send is recognised by. */
  private static final String NOT_SENT_REPORT = "was closed with";

  /**
   * The session a directory server publishes its changes on has no publisher thread: nothing
   * calls {@link Session#start()} on it, so the thread is still {@code NEW} once the broker is
   * connected and has completed its handshake.
   */
  @Test
  public void theSessionOfADirectoryServerBrokerHasNoPublisherThread() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          replicationPort, "sessionPublisherDrainDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
      broker = openReplicationSession(
          baseDN, DS_ID, 100, replicationPort, SOCKET_TIMEOUT_MS, EMPTY_DN_GENID);

      final Session session = sessionOf(broker);
      assertThat(session)
          .as("the broker reported itself connected without a session")
          .isNotNull();
      assertThat(session.isAlive())
          .as("the publisher thread of the session a directory server publishes on is running, "
              + "so publish() enqueues and a close of that session can drop what is queued")
          .isFalse();
      assertThat(session.getState())
          .as("the publisher thread of a broker session was started at some point")
          .isEqualTo(Thread.State.NEW);
    }
    finally
    {
      stop(broker);
      remove(replicationServer);
    }
  }

  /**
   * With no publisher thread, {@code publish()} has written the message to the socket by the
   * time it returns: an immediate close cannot drop it, and the peer reads it after the close.
   */
  @Test
  public void aSessionWithNoPublisherThreadHasReachedTheWireWhenPublishReturns() throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(DS_ID, 0);
    final CSN csn = csns.newCSN();
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        sender.publish(new DeleteMsg(DN.valueOf("uid=onthewire," + TEST_ROOT_DN_STRING),
            csn, "00000000-0000-0000-0000-000000000000"));
        /*
         * No drain, no flush and no wait in between - the close of the restore path of #963 is
         * this close. What publish() already wrote stays readable by the peer: the session
         * closes, it does not unsend the bytes ahead of it.
         */
        sender.close();

        final ReplicationMsg received = receiver.receive();
        assertThat(received)
            .as("the peer did not receive the change published just before the session was closed")
            .isInstanceOf(DeleteMsg.class);
        assertThat(((DeleteMsg) received).getCSN()).isEqualTo(csn);
        /*
         * What makes this the synchronous branch rather than a race won by a publisher thread:
         * there was no publisher thread to win it. The message reached the peer and the session's
         * own thread never ran, so publish() is what wrote it.
         */
        assertThat(sender.getState())
            .as("the session had a publisher thread after all, so the delivery above says only "
                + "that it outran the close, not that publish() wrote the message itself")
            .isEqualTo(Thread.State.NEW);
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * With a publisher thread running, a close sends what that thread had not sent yet rather than
   * dropping it. This is the replication-server end of a session - the end {@code ServerHandler}
   * starts - and the limitation PR #919 recorded.
   * <p>
   * The peer reads the changes and then the {@code StopMsg}, which is what {@link #drain(Session)}
   * stops on: a queue sent after that message would not be counted, so the size below pins the
   * order as well as the delivery.
   */
  @Test
  public void aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed() throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        sender.start();
        sender.waitForStartup();

        /*
         * A reader on the *sending* end, which is what the server keeps running across a close:
         * ServerHandler.shutdown() closes the session (ServerHandler.java:946) and only joins its
         * ServerReader afterwards (:966). It is not decoration here.
         *
         * Without it this end reaches close() with inbound bytes nobody ever read, and a close in
         * that state ends the connection with a reset instead of a FIN - which discards what the
         * peer has not read yet, the whole of what the drain just wrote included. Measured with a
         * bare socket pair, 8 MiB written to a peer reading behind the writer, the only difference
         * between the runs being whether the closing side drained its own inbound:
         *
         *   linux 6.12 / jdk 11    no reader -> 53.4% arrived, "Connection reset"
         *                          reader    -> 100%   arrived, end of stream
         *   macos 15.7 / jdk 26    no reader -> 98.3% arrived, "Connection reset"
         *                          reader    -> 100%   arrived, end of stream
         *
         * Which is why a case without it measures the teardown rather than the drain, and measures
         * it differently per platform: three ubuntu legs of CI lost between 866 and 1694 of these
         * messages while every macos and windows leg passed. This case says so itself - with the
         * start() below commented out and the class run on linux/jdk11, it fails with "the peer
         * received 2873 of the 3000 messages published; the read ended by java.net.SocketException:
         * Connection reset", and passes with it in.
         *
         * Its soTimeout has to go, too: receive() hands a read timeout to setSessionError(), and a
         * session carrying an error skips the drain exactly as it skips the StopMsg.
         */
        sender.setSoTimeout(0);
        final Thread senderReader = newInboundReader(sender);
        senderReader.start();

        /*
         * Nothing reads the peer end while these are published, but that is not what leaves the
         * backlog: 3000 frames of ~150 B are ~440 KB, which the socket buffers of a loopback pair
         * swallow, so the publisher thread is not blocked inside a write. What leaves the backlog
         * is publish() - encode and offer - outrunning the publisher, which writes and flushes one
         * frame at a time through the TLS layer. That is a race rather than a state, so the case
         * asserts below that it was still won when the close ran: with an empty queue there is
         * nothing for the close to drain and this pins nothing.
         */
        for (int i = 0; i < MESSAGES_PUBLISHED; i++)
        {
          sender.publish(new DeleteMsg(DN.valueOf("uid=queued" + i + "," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
        }

        final int queuedAtClose = sendQueueOf(sender).size();
        assertThat(queuedAtClose)
            .as("the publisher had sent everything before the close ran, so this case drained "
                + "nothing: what it measures is the socket rather than close()")
            .isGreaterThan(0);

        final Future<?> closed = executor.submit(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            sender.close();
            return null;
          }
        });

        final Drained drained = drain(receiver);
        closed.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);

        assertThat(drained.received)
            .as("the peer received %d of the %d messages published; the read ended by %s",
                drained.received.size(), MESSAGES_PUBLISHED, drained.endedBy)
            .hasSize(MESSAGES_PUBLISHED);
        assertThat(drained.endedBy)
            .as("the StopMsg is what ended the stream, which is what puts the drained queue "
                + "ahead of it rather than after it")
            .isEqualTo("a StopMsg");
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
    finally
    {
      executor.shutdownNow();
    }
  }

  /**
   * A close which cannot write the queue reports what the peer was not told, once, and counts the
   * message it was writing when the write failed.
   * <p>
   * The road is a peer which is gone by the time the close reaches the queue - a directory server
   * whose {@code StopMsg} brought the {@code ServerReader} of its handler to the {@code close()}
   * of its finally, with the publisher of that session still holding a backlog. Here the sockets
   * of the sender are closed under it instead, which is the same failed write with an exact
   * count: the first {@code send()} of the drain throws, so the report has to name every message
   * the queue held. A peer closed from the outside gives up somewhere inside the TCP buffers
   * instead, and would pin no number at all.
   * <p>
   * The queue is filled through the field rather than by publishing on a started session for the
   * same reason: what a publisher thread has left behind is a race, and this case is the count.
   */
  @Test
  public void aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld() throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        final Queue<byte[]> sendQueue = sendQueueOf(sender);
        for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
        {
          sendQueue.add(new DeleteMsg(DN.valueOf("uid=unsent" + i + "," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000")
              .getBytes(sender.getProtocolVersion()));
        }
        closeTheSocketsUnder(sender);

        final List<String> records = errorLogRecordsOf(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            sender.close();
            return null;
          }
        });

        // Two give-ups are what a drain which does not stop at the first failed write reports.
        final Set<String> reported = reportsIn(records);
        assertThat(reported)
            .as("a close which could not write the queue reports that once, and here it reported: "
                + reported)
            .hasSize(1);
        assertThat(reported.iterator().next())
            .as("the report has to account for the message the failed write took out of the queue "
                + "as well as for the ones left in it")
            .contains(MESSAGES_LEFT_UNSENT + " message(s)")
            .contains("the write failed with");
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * A close of a session which has already failed writes nothing more to it - neither the queue
   * nor the {@code StopMsg} - and still reports the queue it gives up on, once.
   * <p>
   * The error is set through the field, with the sockets left intact: a close which wrote the
   * queue regardless would then get it through, and the peer reading a change is what says so.
   * With the sockets closed under it, as in the case above, such a close would fail at its first
   * write and look the same from the peer.
   */
  @Test
  public void aCloseOfAFailedSessionWritesNothingAndReportsTheQueue() throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        final Queue<byte[]> sendQueue = sendQueueOf(sender);
        for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
        {
          sendQueue.add(new DeleteMsg(DN.valueOf("uid=failed" + i + "," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000")
              .getBytes(sender.getProtocolVersion()));
        }
        final Field sessionError = Session.class.getDeclaredField("sessionError");
        sessionError.setAccessible(true);
        sessionError.set(sender, new java.io.IOException("injected"));

        final List<String> records = errorLogRecordsOf(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            sender.close();
            return null;
          }
        });

        ReplicationMsg read = null;
        try
        {
          read = receiver.receive();
        }
        catch (final java.io.IOException expected)
        {
          // The socket was closed with nothing written to it, which is what this case expects.
        }
        assertThat(read)
            .as("the close wrote to a session which had already failed")
            .isNull();

        final Set<String> reported = reportsIn(records);
        assertThat(reported)
            .as("the close of a failed session holding a queue reports that queue once, and here "
                + "it reported: " + reported)
            .hasSize(1);
        assertThat(reported.iterator().next())
            .contains(MESSAGES_LEFT_UNSENT + " message(s)")
            .contains("the session had already failed")
            .contains("injected");
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * A close of a failed session which has nothing it could not send says nothing: the report is
   * for messages the peer was not told about, and there are none.
   */
  @Test
  public void aCloseOfAFailedSessionWithNothingLeftToSendReportsNothing() throws Exception
  {
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        final Field sessionError = Session.class.getDeclaredField("sessionError");
        sessionError.setAccessible(true);
        sessionError.set(sender, new java.io.IOException("injected"));

        final List<String> records = errorLogRecordsOf(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            sender.close();
            return null;
          }
        });

        assertThat(reportsIn(records))
            .as("the close of a failed session with an empty queue reported a loss")
            .isEmpty();
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * On a started session whose writes fail, the publisher thread goes on taking the queue and
   * failing each write until the close: what it took is lost as much as what it left, and the
   * close reports both. Once closed, the session is off the queueing branch, so a later
   * {@code publish()} - a {@code ServerWriter} or a {@code HeartbeatThread}, which end only on
   * that exception - fails instead of returning as if the message had been queued.
   * <p>
   * The count is exact whatever the thread got through before the close: every message published
   * is either still in the queue or was taken and failed. The case waits for the queue to empty so
   * that it is the publisher's own count the report stands on - the queue alone would report
   * nothing.
   */
  @Test
  public void aCloseOfAStartedSessionWhoseWritesFailedReportsWhatThePublisherTookAsWell()
      throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        sender.start();
        sender.waitForStartup();
        closeTheSocketsUnder(sender);
        for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
        {
          sender.publish(new DeleteMsg(DN.valueOf("uid=takenandlost" + i + ","
              + TEST_ROOT_DN_STRING), csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
        }
        final Queue<byte[]> sendQueue = sendQueueOf(sender);
        final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS);
        while (!sendQueue.isEmpty() && System.nanoTime() - deadline < 0)
        {
          Thread.sleep(10);
        }
        assertThat(sendQueue)
            .as("the publisher thread did not take the queue off a session whose writes fail")
            .isEmpty();

        final List<String> records = errorLogRecordsOf(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            sender.close();
            return null;
          }
        });

        final Set<String> reported = reportsIn(records);
        assertThat(reported)
            .as("the close reports once what the publisher took and could not write, and here "
                + "it reported: " + reported)
            .hasSize(1);
        assertThat(reported.iterator().next())
            .contains(MESSAGES_LEFT_UNSENT + " message(s)")
            .contains("the session had already failed");

        try
        {
          sender.publish(new DeleteMsg(DN.valueOf("uid=afterclose," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
          org.assertj.core.api.Assertions.fail(
              "a publish() on a closed session returned as if it had queued the message");
        }
        catch (final java.io.IOException expected)
        {
          // The close put the session on the synchronous branch, which fails on the closed socket.
        }
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * A session closed before its publisher thread is started - which a handler whose session is
   * closed by a stop of all servers during its handshake does, before it goes on to start it -
   * stays on the synchronous branch of {@code publish()}. Put on the queueing one, a publish
   * would return at once, as if queued, onto a queue nothing sends, and a {@code HeartbeatThread}
   * started after it would go on publishing into it until the server stops.
   */
  @Test
  public void aSessionClosedBeforeItIsStartedKeepsFailingPublishes() throws Exception
  {
    final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session sender = pair[0];
      final Session receiver = pair[1];
      try
      {
        sender.close();
        sender.start();
        sender.waitForStartup();
        sender.join(SOCKET_TIMEOUT_MS);

        try
        {
          sender.publish(new DeleteMsg(DN.valueOf("uid=afterclose," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
          org.assertj.core.api.Assertions.fail(
              "a publish() on a session closed before it was started returned as if it had "
              + "queued the message");
        }
        catch (final java.io.IOException expected)
        {
          // The synchronous branch, failing on the closed socket.
        }
      }
      finally
      {
        StaticUtils.close(sender, receiver);
      }
    }
  }

  /**
   * The reports of a queue not sent among the records, from the text of the report on, so that
   * the severity and the timestamp a record carries do not make one report captured twice look
   * like two - and so that two reports still do.
   */
  private static Set<String> reportsIn(final List<String> records)
  {
    final Set<String> reported = new LinkedHashSet<>();
    for (final String record : records)
    {
      final int start = record.indexOf(NOT_SENT_REPORT);
      if (start >= 0)
      {
        reported.add(record.substring(start));
      }
    }
    return reported;
  }

  /**
   * Consumes whatever arrives on a session until it is closed, as {@code ServerReader} does for a
   * server handler.
   * <p>
   * It is what it reads at the socket rather than what it returns that matters: the peer of these
   * cases publishes nothing, so this returns no message at all, while the read it sits in is what
   * keeps the receive queue of this end empty - which is the condition a close needs to end the
   * connection in an orderly way.
   */
  private Thread newInboundReader(final Session session)
  {
    final Thread reader = new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          while (true)
          {
            session.receive();
          }
        }
        catch (final Exception ignored)
        {
          // The close of the session ends the read, which is the end of this thread.
        }
      }
    }, "inbound reader of " + session.getName());
    reader.setDaemon(true);
    return reader;
  }

  /**
   * Reads until the session gives nothing back, and answers the CSNs of the changes it read
   * together with what ended the read.
   * <p>
   * Why the reason is carried rather than swallowed: a short read is exactly the failure this
   * suite is about, and "the peer received fewer than were sent" does not say whether the stream
   * ended orderly at a {@code StopMsg} or was cut off - which are different defects.
   */
  private static final class Drained
  {
    private final List<CSN> received = new CopyOnWriteArrayList<>();
    private String endedBy;

    @Override
    public String toString()
    {
      return received.size() + " change(s), ended by " + endedBy;
    }
  }

  private Drained drain(final Session session)
  {
    final Drained drained = new Drained();
    try
    {
      while (true)
      {
        final ReplicationMsg msg = session.receive();
        if (msg instanceof DeleteMsg)
        {
          drained.received.add(((DeleteMsg) msg).getCSN());
        }
        else if (msg instanceof StopMsg)
        {
          drained.endedBy = "a StopMsg";
          return drained;
        }
        else
        {
          drained.endedBy = "an unexpected " + msg.getClass().getSimpleName();
          return drained;
        }
      }
    }
    catch (final Exception e)
    {
      drained.endedBy = e.getClass().getName() + ": " + e.getMessage();
      return drained;
    }
  }

  /** The queue a started session's publisher thread takes its buffers from. */
  @SuppressWarnings("unchecked")
  private static Queue<byte[]> sendQueueOf(final Session session) throws Exception
  {
    final Field sendQueue = Session.class.getDeclaredField("sendQueue");
    sendQueue.setAccessible(true);
    return (Queue<byte[]>) sendQueue.get(session);
  }

  /**
   * Closes the sockets a session writes through while leaving the session unaware of it, so that
   * its next write fails - the state a peer which has gone away leaves it in.
   */
  private static void closeTheSocketsUnder(final Session session) throws Exception
  {
    for (final String name : new String[] { "secureSocket", "plainSocket" })
    {
      final Field socket = Session.class.getDeclaredField(name);
      socket.setAccessible(true);
      StaticUtils.close((Closeable) socket.get(session));
    }
  }

  /** The session a broker publishes on, which it keeps to itself. */
  private Session sessionOf(final ReplicationBroker broker) throws Exception
  {
    final Field connectedRSField = ReplicationBroker.class.getDeclaredField("connectedRS");
    connectedRSField.setAccessible(true);
    final Object connectedRS = ((AtomicReference<?>) connectedRSField.get(broker)).get();
    final Field sessionField = connectedRS.getClass().getDeclaredField("session");
    sessionField.setAccessible(true);
    return (Session) sessionField.get(connectedRS);
  }

  /**
   * A connected pair of sessions over the loopback, the client end first. Neither end is
   * started: a session which publishes synchronously is what a broker has, and the test which
   * needs a publisher thread starts the end it needs.
   */
  private Session[] connectSessionPair(final ServerSocket listen) throws Exception
  {
    final ReplSessionSecurity security = getReplSessionSecurity();
    final Socket clientSocket = new Socket("127.0.0.1", listen.getLocalPort());
    clientSocket.setTcpNoDelay(true);
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    try
    {
      final Future<Session> clientEnd = executor.submit(new Callable<Session>()
      {
        @Override
        public Session call() throws Exception
        {
          return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS);
        }
      });
      final Socket serverSocket = listen.accept();
      serverSocket.setTcpNoDelay(true);
      final Session serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS);
      return new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
    }
    finally
    {
      executor.shutdown();
    }
  }
}

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

@@ -726,9 +726,13 @@
          .isNotNull();
      /*
       * The forward asserted above proves the message reached the Session, not the wire: close()
       * discards whatever is still in its send queue without draining it, which is the
       * limitation issue #919 recorded. If this is the only assertion which fails, that window
       * is the explanation rather than the granularity of the barrier.
       * now sends what its publisher left queued, but only within its own budget, and it writes
       * that queue under publishLock so that a message published meanwhile lands after it rather
       * than between two of its own. If this is the only assertion which fails, the close is
       * where to look before the granularity of the barrier: the warning close() writes for a
       * queue it could not hand over says why it gave that queue up, and its absence does not
       * prove the message left this end - a publish() concurrent with the close is still dropped
       * at the door without one.
       */
      assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the peer which was held back never learned that the replica went offline, "