[#911] Report the replication connections which used to be dropped in silen...

Valery Kharseko

23 hours ago 6477a7e28301b6d5d9746cf4bfb29ae4b748258d

[#911] Report the replication connections which used to be dropped in silence (#935)

 opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-troubleshooting.adoc

@@ -781,10 +781,25 @@
 are misconfigured: check that the certificate nickname configured in the
 crypto manager exists in the ads-truststore, and that every server of the
 topology trusts the certificates of the other servers. At most one such
 failure is logged as a warning every 5 minutes; the others go to the debug
 log, which is disabled by default (0 since the previous warning).
 failure is logged as a warning every 5 minutes; the others are recorded with
 the information severity, which the replication log publishes and the error
 log does not (0 since the previous warning).
 The error was: Remote host closed connection during handshake
----
A connection can also fail before it becomes a replication session for a reason which is not a failed handshake, and further messages report that:

* `Replication server RS(1) accepted a connection from ... but could not start a replication session on it, and closed it` reports an inbound connection dropped for such a reason. An `ads-truststore` which was deleted, truncated, or made unreadable after the server started fails every inbound connection this way: the file is read again for each of them, so a server which is running is not a server whose trust store is still there.

* `Replication server RS(1) could not accept a connection on ...` reports a failure of the listen socket itself, the process running out of file descriptors being the usual one. When such a failure repeats at once, the listen thread waits 100 ms before accepting again, so that it does not spin on it.

* `Replication server RS(1) could not connect to replication server ... for domain ...` reports the outgoing half, which used to be visible only in the log of the peer, that is on the machine whose configuration is right. It is logged once for each outage, and the end of the outage is reported by one of two messages, so a replication server stopped for maintenance costs two lines rather than one every few seconds. `Replication server RS(1) connected to replication server ...` reports a peer which answered and completed the handshake, which is a replication session. `Replication server RS(1) reached replication server ... but the handshake with it did not complete` reports a peer which answered on its replication port while the handshake did not complete: the outage is over, and there is still nothing replicating over that connection. Where this server ended the handshake, whether it rejected the peer or the peer went away while the handshake was running, the reason is logged next to that message; where nothing is logged next to it, the handshake was ended by the peer, which may have logged no reason of its own either -- a shutdown under way, or a connection the two servers made to each other at the same time, of which one is dropped while the other one serves the domain. Two replication servers which share a server id are reported by the reason logged for the handshake, which is logged whether or not the outgoing message above accompanies it: that message follows an outage which was reported, and two servers which have been reachable since they started never opened one. A session established afterwards is reported in its turn. A session lost later is not reported by these messages: the loss is logged where the reader of that session ends, and the next of these comes when the peer can no longer be reached at all. A replication server which cannot be reached, or whose handshake does not complete, is left alone for a few passes of the connect thread before it is tried again, and it is left alone for every domain: the domain named is the one the failed attempt was for rather than the only one affected. The domain whose attempt failed is also the one retried, so for as long as the condition lasts the other domains of that peer are not dialled from this server again -- they are served by the connection the peer makes to this server, replication servers dialling each other both ways, and only where the peer is up to make it: a peer which is down serves none of its domains anyway.

* On the directory server side, the reason each replication server could not be used is now logged for every one of them, before the `was unable to connect to any replication servers for domain` summary which names none: `Please check that there is a replication server listening at this address` for a refused connection, `timed out while connecting to replication server` for an unanswered one, and `encountered an unexpected error while connecting to replication server` followed by the cause for the rest, a rejected certificate included. Only the summary used to be logged when no replication server answered at all.

The failed handshake above and the first of these are reported once for every connection which reaches the replication port, and the second once for every failed `accept()`, which on a process out of file descriptors is as often as the listen loop can turn. Each of the three is therefore logged as a warning at most once every five minutes, and each warning reports how many failures it stands for.

The failures those warnings suppress are still recorded, with the `information` severity. `logs/replication` publishes that severity for replication messages, so the suppressed records are there rather than nowhere; `logs/errors` does not publish it. To have them in the error log as well, add `info`, which is the name that severity goes by in the configuration, to the `default-severity` property of the error log publisher. They do not go to the debug log.

OpenDJ maintains historical information about changes in order to bring replicas up to date, and to resolve replication conflicts. To prevent historical information from growing without limit, OpenDJ purges historical information after a configurable delay (`replication-purge-delay`, default: 3 days). A replica can become irrevocably out of sync if you restore it from a backup archive older than the purge delay, or if you stop it for longer than the purge delay. If this happens to you, disable the replica, and then reinitialize it from a recent backup or from a server that is up to date.



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

@@ -24,7 +24,6 @@
import java.net.Socket;
import java.util.SortedSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLException;
@@ -35,6 +34,7 @@
import org.forgerock.opendj.config.server.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.types.CryptoManager;
import org.opends.server.util.FailureLogThrottle;

/**
 * This class represents the security configuration for replication protocol
@@ -55,7 +55,8 @@
   * Minimum interval, in minutes, between two warnings about a failed SSL handshake
   * on the replication port. Every connection which is not a replication peer fails
   * the handshake, network probes included, so only the first failure of an interval
   * is logged as a warning and the following ones are logged at debug level.
   * is logged as a warning and the following ones are recorded with the information
   * severity, which the replication log publishes and the error log does not.
   */
  private static final long HANDSHAKE_FAILURE_WARN_INTERVAL_MINUTES = 5;

@@ -63,16 +64,9 @@
  static final long HANDSHAKE_FAILURE_WARN_INTERVAL_NANOS =
      TimeUnit.MINUTES.toNanos(HANDSHAKE_FAILURE_WARN_INTERVAL_MINUTES);

  /**
   * Value of {@link System#nanoTime()} at which the last handshake failure was
   * logged as a warning. It starts one interval in the past so that the first
   * failure is warned about.
   */
  private final AtomicLong lastHandshakeFailureWarnNanos =
      new AtomicLong(System.nanoTime() - HANDSHAKE_FAILURE_WARN_INTERVAL_NANOS);

  /** Number of handshake failures logged at debug level since the last warning. */
  private final AtomicLong suppressedHandshakeFailures = new AtomicLong();
  /** Bounds how often a failed handshake is warned about. */
  private final FailureLogThrottle handshakeFailures =
      new FailureLogThrottle(HANDSHAKE_FAILURE_WARN_INTERVAL_MINUTES, TimeUnit.MINUTES);

  /**
   * Whether replication sessions use SSL encryption.
@@ -300,8 +294,9 @@
  /**
   * Logs a failed SSL handshake on the replication port, as a warning for the
   * first failure of each {@link #HANDSHAKE_FAILURE_WARN_INTERVAL_MINUTES}
   * interval and at debug level for the following ones. The warning reports how
   * many failures were logged at debug level before it, so that a single line
   * interval and with the information severity for the following ones, which the
   * replication log publishes and the error log does not. The warning reports how
   * many failures were recorded that way before it, so that a single line
   * cannot be mistaken for a single failed connection. That count looks backwards
   * only: the failures which follow the last warning of a burst are counted but
   * never reported, as nothing flushes the count when the failures stop.
@@ -330,28 +325,22 @@

  /**
   * Records a handshake failure which happened at the provided time and tells how it
   * must be logged, together with the number of failures logged at debug level since
   * the previous warning.
   * must be logged, together with the number of failures suppressed since the previous
   * warning.
   * <p>
   * Package private for testing.
   *
   * @param nowNanos
   *          The value of {@link System#nanoTime()} at which the handshake failed.
   * @return A number greater than or equal to zero if this failure is to be logged as a
   *         warning, which is then the number of failures logged at debug level since
   *         the previous warning, or {@code -count - 1} if this failure is itself to be
   *         logged at debug level, where {@code count} is the number of failures logged
   *         at debug level since the previous warning, this one included.
   *         warning, which is then the number of failures suppressed since the previous
   *         warning, or {@code -count - 1} if this failure is itself to be suppressed,
   *         where {@code count} is the number of failures suppressed since the previous
   *         warning, this one included.
   */
  long recordHandshakeFailure(final long nowNanos)
  {
    final long lastWarn = lastHandshakeFailureWarnNanos.get();
    if (nowNanos - lastWarn >= HANDSHAKE_FAILURE_WARN_INTERVAL_NANOS
        && lastHandshakeFailureWarnNanos.compareAndSet(lastWarn, nowNanos))
    {
      return suppressedHandshakeFailures.getAndSet(0);
    }
    return -suppressedHandshakeFailures.incrementAndGet() - 1;
    return handshakeFailures.record(nowNanos);
  }

  /**

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

@@ -37,12 +37,16 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizableMessageBuilder;
import org.forgerock.i18n.LocalizableMessageDescriptor;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.config.server.ConfigException;
@@ -81,6 +85,7 @@
import org.opends.server.types.HostPort;
import org.opends.server.types.SearchFilter;
import org.opends.server.types.VirtualAttributeRule;
import org.opends.server.util.FailureLogThrottle;

/**
 * ReplicationServer Listener. This singleton is the main object of the
@@ -135,6 +140,31 @@

  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /**
   * Minimum interval, in minutes, between two warnings about the same failure of the
   * listen thread. Every connection reaching the replication port goes through that
   * thread, so the rate has to be bounded. Each message reports the interval it was
   * logged under, so this one is free to differ from the one a failed handshake is
   * reported with, which happens to be the same five minutes.
   */
  private static final long FAILURE_WARN_INTERVAL_MINUTES = 5;

  /**
   * Time, in milliseconds, the listen thread waits before accepting again after
   * {@link ServerSocket#accept()} failed.
   * <p>
   * Package private for testing: a test which makes {@code accept()} fail asserts that
   * the thread waited instead of spinning, so the wait has to be a value it can name.
   */
  static final long ACCEPT_FAILURE_BACKOFF_MS = 100;

  /** {@link #ACCEPT_FAILURE_BACKOFF_MS}, which is also how close two failures have to be to count as repeating. */
  private static final long ACCEPT_FAILURE_BACKOFF_NANOS =
      TimeUnit.MILLISECONDS.toNanos(ACCEPT_FAILURE_BACKOFF_MS);

  /** Reports a peer this replication server cannot connect to once per outage. */
  private final ConnectFailureReporter connectFailures = new ConnectFailureReporter();

  /** To know whether a domain is enabled for the external changelog. */
  private final ECLEnabledDomainPredicate domainPredicate;

@@ -288,18 +318,75 @@
        socket.getInetAddress().getHostAddress(),
        socket.getLocalPort());

    /*
     * When the last failure of accept() on this socket was done being handled, confined to
     * this thread: a listen port change runs a second listen thread, and each of them times
     * the socket it accepts on.
     *
     * A failure which follows one closely is what the wait below is for. It starts a whole
     * interval in the past, so an isolated failure does not wait, and a successful accept
     * puts it back there: a loop which is serving connections is doing work rather than
     * spinning, and what paced it before is not what the next failure repeats. Without that
     * reset, a stream of connections aborted between the handshake and accept() -- a health
     * check or a port scan -- would charge the whole listen port a wait per probe, and the
     * peers queued behind them would pay it.
     *
     * What the reset gives up is the failure which alternates with a connection: a process
     * out of file descriptors frees one now and then, the accept it lets through resets the
     * clock, and the failure after it is timed as isolated and not waited on. The loop then
     * turns as fast as the connections arrive. That is the trade -- a probe must not cost
     * the peers behind it a wait, and a loop which is accepting is not the silent spin the
     * wait was added for -- and the warning is throttled either way, so the error log holds
     * one line per five minutes of it whichever side of the trade the failure falls on.
     *
     * Read when the previous failure was handled rather than when it happened, so that the
     * wait it was granted is not what makes the next failure look isolated: timing from the
     * failure would leave every second one of a continuous run unwaited, and bound the spin
     * to twice the rate the wait is chosen for.
     */
    long handledAcceptFailureNanos = System.nanoTime() - ACCEPT_FAILURE_BACKOFF_NANOS;

    /*
     * Bound how often a failure of accept() on this socket, and a connection accepted on it
     * which cannot be turned into a session, are warned about. Both are confined to this
     * thread for the same reason the clock above is: a listen port change runs a second
     * listen thread -- switchListenPort() starts it before it stops this one -- and a five
     * minute window opened on the port which was left would otherwise suppress the first
     * failure on the port which replaced it, silence right after the administrator changed
     * the port to get out of trouble.
     */
    final FailureLogThrottle acceptFailures =
        new FailureLogThrottle(FAILURE_WARN_INTERVAL_MINUTES, TimeUnit.MINUTES);
    final FailureLogThrottle sessionSetupFailures =
        new FailureLogThrottle(FAILURE_WARN_INTERVAL_MINUTES, TimeUnit.MINUTES);

    while (!shutdown.get() && !socket.isClosed())
    {
      // Wait on the replicationServer port.
      // Read incoming messages and create LDAP or ReplicationServer listener
      // and Publisher.
      Session session = null;
      try
      {
        Session session;
        Socket newSocket = null;
        final Socket newSocket;
        try
        {
          newSocket = socket.accept();
        }
        catch (Exception e)
        {
          final boolean repeated =
              System.nanoTime() - handledAcceptFailureNanos < ACCEPT_FAILURE_BACKOFF_NANOS;
          handleAcceptFailure(acceptFailures, socket, e, repeated);
          handledAcceptFailureNanos = System.nanoTime();
          continue;
        }
        // A connection served is not a spin: the failures before it stop pacing the loop,
        // and a failure alternating with a connection is therefore never waited on.
        handledAcceptFailureNanos = System.nanoTime() - ACCEPT_FAILURE_BACKOFF_NANOS;

        try
        {
          newSocket.setTcpNoDelay(true);
          newSocket.setKeepAlive(true);
          int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
@@ -311,12 +398,13 @@
        }
        catch (Exception e)
        {
          // If problems happen during the SSL handshake, it is necessary
          // to close the socket to free the associated resources.
          if (newSocket != null)
          {
            newSocket.close();
          }
          logSessionSetupFailure(sessionSetupFailures, newSocket, e);
          // createServerSession() closes the socket itself when it does not return a
          // session, so this closes the one whose options could not be set, and closes a
          // second time, harmlessly, the one it already released. Through the helper: a
          // close which throws here would escape this catch into the one of the loop,
          // which reports it as a failure to listen, without a throttle.
          close(newSocket);
          continue;
        }

@@ -349,6 +437,16 @@
      }
      catch (Exception e)
      {
        /*
         * A session which reaches here is owned by nothing else: both handlers started
         * above abort a handshake they cannot complete themselves, closing the session
         * and returning, so what lands here is the session of a peer which failed after
         * its handshake -- receive() on a peer which was killed, or which stopped
         * answering past the connection timeout. Leaving it open leaked the socket and
         * its file descriptor for the life of the process, one per such peer, towards the
         * very exhaustion the accept loop above now has to survive.
         */
        close(session);
        // The socket has probably been closed as part of the
        // shutdown or changing the port number process.
        // Just log debug information and loop.
@@ -363,6 +461,128 @@
  }

  /**
   * Reports a failure of {@link ServerSocket#accept()} and, when it is not the first one
   * in a row, waits before accepting again.
   * <p>
   * The listen socket stays open across such a failure, so the loop would come straight
   * back to {@code accept()} and, when the cause is the process running out of file
   * descriptors, spin on it without ever logging anything. The wait is what bounds that
   * spin; the throttle is what bounds the log.
   * <p>
   * Only a failure which repeats is waited on: a connection reset between the handshake
   * and {@code accept()} fails it once, and making the listen thread pause for a
   * connection nobody is waiting for any more would slow down the ones which follow it.
   *
   * @param throttle
   *          The throttle of the calling listen thread, which bounds how often this
   *          failure is warned about.
   * @param socket
   *          The socket connections are accepted on.
   * @param e
   *          The failure.
   * @param repeated
   *          Whether the previous failure of {@code accept()} on that socket was handled
   *          less than {@link #ACCEPT_FAILURE_BACKOFF_MS} ago.
   */
  private void handleAcceptFailure(final FailureLogThrottle throttle, final ServerSocket socket,
      final Exception e, final boolean repeated)
  {
    // Read before the socket is tested rather than after it is reported: a socket closed
    // in between is one this method returns on, while a closed socket names no address at
    // all, and the warning would report the port the administrator needs as "null".
    final Object listenAddress = socket.getLocalSocketAddress();
    if (shutdown.get() || socket.isClosed())
    {
      // The socket was closed to stop this thread or to change the listen port: the loop
      // is about to end, and this failure is how it is told to.
      logger.traceException(e);
      return;
    }

    logThrottledFailure(throttle, WARN_REPLICATION_SERVER_ACCEPT_ERROR, listenAddress, e);
    if (repeated)
    {
      // The wait is not interruptible on purpose: the flag is left alone, as restoring it
      // would make every wait which follows return at once and bring the spin back.
      // Nothing is lost by that -- the only interrupt the listen thread gets, from
      // abortInitialization(), comes after the shutdown flag is set and the socket
      // closed, which is what ends this loop.
      sleep(ACCEPT_FAILURE_BACKOFF_MS);
    }
  }

  /**
   * Reports a connection which was accepted but on which no replication session could be
   * started, and which is therefore about to be closed.
   * <p>
   * A failed SSL handshake is reported by {@link ReplSessionSecurity#createServerSession}
   * itself, so what reaches here is everything else: a trust store which cannot be read,
   * which makes every inbound connection fail this way, and a peer which stops
   * responding during the handshake. Every connection reaching the replication port goes
   * through this path, so the log has to be throttled the way the handshake failure next
   * door is.
   *
   * @param throttle
   *          The throttle of the listen thread which accepted the connection, which is
   *          confined to it: see where it is built.
   * @param socket
   *          The accepted socket, which {@code createServerSession} has usually closed
   *          already, in its own {@code finally}: a closed socket still names the peer it
   *          was connected to, which is what this reads from it.
   * @param e
   *          The failure.
   */
  private void logSessionSetupFailure(final FailureLogThrottle throttle, final Socket socket,
      final Exception e)
  {
    logThrottledFailure(throttle, WARN_REPLICATION_SERVER_SESSION_SETUP_ERROR,
        socket.getRemoteSocketAddress(), e);
  }

  /**
   * Logs a failure of the listen thread as a warning when the provided throttle lets it
   * through, reporting how many failures that line stands for.
   * <p>
   * A failure the throttle suppresses is still recorded, with the information severity:
   * {@code logger.debug} of a localized message publishes to the error log and not to the
   * debug log. What reads that severity is the replication log, whose shipped publisher
   * overrides it on for the {@code SYNC} category; the error log does not publish it by
   * default. So the throttle bounds what the error log holds, and what the replication log
   * holds is one record per failure, which is what the messages say.
   *
   * @param throttle
   *          The throttle bounding how often this failure is warned about.
   * @param message
   *          The message reporting it, which takes the server id, the address, the cause,
   *          the interval and the number of failures suppressed since the previous warning.
   * @param address
   *          The address the failure happened on.
   * @param e
   *          The failure.
   */
  private void logThrottledFailure(final FailureLogThrottle throttle,
      final LocalizableMessageDescriptor.Arg5<Number, Object, Object, Number, Number> message,
      final Object address, final Exception e)
  {
    // The stack trace goes to the trace log, which formats it only when it is enabled.
    logger.traceException(e);

    final long recorded = throttle.record(System.nanoTime());
    // Not a stack trace: the arguments of a message are built whether or not the record
    // they go into is published, and this one is written in a state the server has to
    // survive.
    final LocalizableMessage cause = getExceptionMessage(e);
    if (recorded >= 0)
    {
      logger.warn(message, getServerId(), address, cause, FAILURE_WARN_INTERVAL_MINUTES, recorded);
    }
    else
    {
      logger.debug(message, getServerId(), address, cause, FAILURE_WARN_INTERVAL_MINUTES, -recorded - 1);
    }
  }

  /**
   * This method manages the connection with the other replication servers.
   * It periodically checks that this replication server is indeed connected
   * to all the other replication servers and if not attempts to
@@ -376,8 +596,11 @@
      while (!shutdown.get())
      {
        HostPort localAddress = HostPort.localAddress(getReplicationPort());
        final Set<HostPort> configuredRSAddresses = getConfiguredRSAddresses();
        final Set<DN> domainBaseDNs = new HashSet<>();
        for (ReplicationServerDomain domain : getReplicationServerDomains())
        {
          domainBaseDNs.add(domain.getBaseDN());
          /*
           * If there are N RSs configured then we will usually be connected to
           * N-1 of them, since one of them is usually this RS. However, we
@@ -386,11 +609,17 @@
           */
          final Set<HostPort> connectedRSAddresses =
              getConnectedRSAddresses(domain);
          for (HostPort rsAddress : getConfiguredRSAddresses())
          for (HostPort rsAddress : configuredRSAddresses)
          {
            if (connectedRSAddresses.contains(rsAddress))
            {
              continue; // Skip: already connected.
              // Skip: already connected. The connection may be the one that peer made to
              // this server, which connect() never sees, so this is where a failure
              // reported for a peer which came back on its own is closed: leaving it
              // recorded would silence the next outage of that peer. A handler registered
              // with the domain is a handshake which completed, so this reports a session.
              reportConnectionRestored(rsAddress, domain.getBaseDN(), true);
              continue;
            }

            // FIXME: this will need changing if we ever support listening on
@@ -413,6 +642,12 @@
          }
        }

        // A peer or a domain which is no longer configured is never connected to again,
        // so nothing would ever clear a failure recorded for it. Forget those: a peer
        // taken out of the configuration while it is down, and put back while it still
        // is, would otherwise have its first failure silenced.
        connectFailures.retainAll(configuredRSAddresses, domainBaseDNs);

        // Notify any threads waiting with domain tickets after each iteration.
        synchronized (domainTicketLock)
        {
@@ -448,13 +683,21 @@

  /**
   * Establish a connection to the server with the address and port.
   * <p>
   * Package private for testing: what a peer costs the error log is decided here, and a
   * test of the bookkeeping alone cannot see how it is called.
   *
   * @param remoteServerAddress
   *          The address and port for the server
   * @param baseDN
   *          The baseDN of the connection
   * @return {@code true} if the peer is connected to, {@code false} if it could not be
   *         reached at all or if the handshake offered to it did not complete. The
   *         caller leaves a peer it gets {@code false} for alone for a few passes, which
   *         a peer answering with an abort every second is as much in need of as one
   *         which does not answer.
   */
  private boolean connect(HostPort remoteServerAddress, DN baseDN)
  boolean connect(HostPort remoteServerAddress, DN baseDN)
  {
    boolean sslEncryption = replSessionSecurity.isSslEncryption();

@@ -466,6 +709,7 @@

    Socket socket = new Socket();
    Session session = null;
    final boolean handshakeCompleted;
    try
    {
      socket.setReuseAddress(true);
@@ -490,16 +734,250 @@

      ReplicationServerHandler rsHandler = new ReplicationServerHandler(
          session, config.getQueueSize(), this, config.getWindowSize());
      rsHandler.connect(baseDN, sslEncryption);
      handshakeCompleted = rsHandler.connect(baseDN, sslEncryption);
    }
    catch (Exception e)
    {
      logger.traceException(e);
      if (connectFailures.recordFailure(remoteServerAddress, baseDN))
      {
        // The failure used to be traced and nothing else, so a replication server whose
        // outgoing handshake failed logged nothing at all: the only log naming the problem
        // was the one of the peer, which is the machine whose configuration is right.
        logger.warn(WARN_REPLICATION_SERVER_CONNECT_ERROR, getServerId(), remoteServerAddress,
            baseDN, getExceptionMessage(e));
      }
      close(session);
      close(socket);
      return false;
    }
    return true;
    /*
     * The outage closed here is a failure to connect, and reaching this line is the peer
     * answering on its replication port: everything WARN_REPLICATION_SERVER_CONNECT_ERROR
     * is reported for is above it -- the socket and the session built on it, the handshake
     * throwing nothing of its own. So the outage is closed whatever the handshake did next,
     * and what the handshake did next decides which recovery is reported rather than
     * whether one is.
     *
     * Closing it on the connection alone is what keeps the peers this server never sees
     * connected under the address it dialled reportable, and there is one of those for each
     * narrower reading:
     *
     * the registration with the domain misses a peer which negotiates protocol version 1.
     * ReplicationServerHandler.connect() registers only above V1, the FIXME there being
     * older than this, so such a peer is connected and never registered.
     *
     * the address, and the session left open with it, miss a peer which dials out from an
     * address other than the one it is configured under -- multi homing, NAT. Its inbound
     * handler is registered under the source address of its own connection,
     * ServerHandler.toServerAddressURL() reading the host from the session, so the already
     * connected branch of runConnect() compares the configured address against one it never
     * matches, and the handshake this server offers that same peer aborts on a duplicate
     * server id: abortStart() closes the session, and an open session is never seen here
     * again.
     *
     * An outage left open is not a line too few but a peer gone silent: recordFailure()
     * returns false from then on, so the next real outage of it is not reported at all.
     *
     * What the handshake did next is reported all the same, because it is not the log which
     * can be left to it. ReplicationServerHandler.connect() aborts at seven places, and the
     * three which pass no message log nothing at all, abortStart() logging nothing without
     * one: a StopMsg read where the peer's ReplServerStartMsg was due, a cross connect this
     * server resolves against a peer it is already connected to, and a phase two the peer
     * leaves unanswered. Those would otherwise leave "connected" as the last thing said
     * about a domain which has no session for it. The remaining four have their reason
     * logged here by abortStart(), and they are not all of one kind: this server rejecting
     * the peer, the peer going away while the handshake ran, and whatever else fails on
     * this side of it. So the message reported for an answer without a session tells the
     * two apart by where a reason was logged rather than by naming a side: the operator is
     * sent to the log of the peer only where nothing was logged here, and that is also the
     * abort the peer may have logged nothing about either -- it resolves a cross connect
     * with the same silent abortStart(null), one line up in startFromRemoteRS().
     *
     * The cross connect this server resolves is the one abort of the three where a session
     * for the domain does exist: it is the connection the peer made, which the already
     * connected branch of runConnect() reports on its next pass. Reaching this line with
     * one open needs that registration to land between the snapshot that branch reads and
     * the dial below it, so what it costs is one warning, and the pass after it says what
     * is true.
     */
    reportConnectionRestored(remoteServerAddress, baseDN, handshakeCompleted);
    return handshakeCompleted;
  }

  /**
   * Reports that a peer this replication server had reported it could not connect to can be
   * reached again, and does nothing when what is now true of that peer is what was last
   * reported about it.
   * <p>
   * A peer which answers is no longer the peer the outage was reported for, whether the
   * handshake completed or not: leaving the outage recorded for one which answers and aborts
   * every handshake would silence its next real outage. But the two are not the same
   * recovery, and the difference is one the operator has to be able to read -- a peer which
   * stops the handshake is one this server can reach and has no session with, which is not
   * what "connected" says -- so each moves the peer to a state of its own rather than
   * clearing what is known about it. That is what leaves the session established after an
   * abort still reportable: it is a recovery from the answer without a session, which the
   * clearing form had already consumed.
   *
   * @param remoteServerAddress
   *          The address of the peer which answered.
   * @param baseDN
   *          The base DN of the domain the attempt was for.
   * @param connected
   *          Whether the handshake completed, so that this server is connected to the peer,
   *          rather than aborted after it had answered.
   */
  private void reportConnectionRestored(final HostPort remoteServerAddress, final DN baseDN,
      final boolean connected)
  {
    if (connected)
    {
      if (connectFailures.recordConnected(remoteServerAddress, baseDN))
      {
        logger.info(NOTE_REPLICATION_SERVER_CONNECT_RESTORED, getServerId(), remoteServerAddress, baseDN);
      }
    }
    else if (connectFailures.recordReachableWithoutSession(remoteServerAddress, baseDN))
    {
      logger.warn(WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION, getServerId(), remoteServerAddress, baseDN);
    }
  }

  /**
   * Remembers which replication servers this replication server has already reported it
   * cannot connect to, so that a peer which stays unreachable is reported once instead of
   * on every attempt to reach it.
   * <p>
   * The connect thread retries a failed peer every few seconds, for as long as it is down,
   * and a peer being down is a normal state: one stopped for maintenance would otherwise
   * fill the error log for the duration. The failure is reported when it starts and, through
   * {@link #recordConnected} and {@link #recordReachableWithoutSession}, when it ends, so
   * that neither end of it has to be inferred from a silence.
   * <p>
   * What is held per peer and domain is the last state which was <em>reported</em>, not
   * whether an outage is open. A record which is merely present or absent cannot carry the
   * middle state -- a peer which answers on its replication port and stops the handshake --
   * because reporting it would consume the record, and the session which is established
   * seconds later would then find nothing left to close and go unreported. A peer restarting
   * takes exactly that path: its port answers before its domains are up.
   * <p>
   * Package private for testing.
   */
  static final class ConnectFailureReporter
  {
    /** What the last message about a peer and a domain said about them. */
    private enum Reported
    {
      /** The peer could not be reached at all: {@code WARN_REPLICATION_SERVER_CONNECT_ERROR}. */
      DOWN,
      /**
       * The peer answered and the handshake did not complete:
       * {@code WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION}.
       */
      REACHABLE_NO_SESSION;
    }

    /**
     * What was last reported about each domain of each peer, holding no entry for the peers
     * and domains nothing was reported about, which a connection to them is therefore not to
     * be reported for.
     * <p>
     * Keyed by {@link HostPort} rather than by its string form, which is the raw host
     * while equality is on the normalized one: two spellings of the same peer would
     * otherwise be two keys, and a failure recorded under one would never be cleared by
     * the connection recorded under the other.
     * <p>
     * A HostPort holds the resolution its host had when it was built, and is documented as
     * not meant to be cached. {@link #retainAll} is what makes holding one here bounded: it
     * runs on every pass of the connect thread against freshly built addresses, so a key
     * whose resolution has drifted is dropped within one pass, and the outage under it is
     * reported a second time rather than never reported again.
     * <p>
     * The connect thread is the only one which connects to peers, but a configuration
     * change replaces it, so this is a map two threads may hand over.
     */
    private final ConcurrentMap<HostPort, ConcurrentMap<DN, Reported>> reported =
        new ConcurrentHashMap<>();

    /**
     * Records a failure to connect to the provided peer for the provided domain.
     *
     * @param peer
     *          The address of the replication server which could not be connected to.
     * @param baseDN
     *          The base DN of the domain the connection was for.
     * @return {@code true} if this failure is to be reported, {@code false} if that peer is
     *         already known to be unreachable for that domain.
     */
    boolean recordFailure(final HostPort peer, final DN baseDN)
    {
      return Reported.DOWN != domainsOf(peer).put(baseDN, Reported.DOWN);
    }

    /**
     * Records that the provided peer answered on its replication port for the provided
     * domain and that the handshake did not complete, so that this server has no session
     * with it.
     * <p>
     * Reported only where an outage was: an abort which carries a reason logs that reason on
     * this server, and one which does not is the peer's to log. What this closes is the
     * message which said the peer could not be reached, which is no longer what is wrong
     * with it.
     *
     * @param peer
     *          The address of the replication server which answered.
     * @param baseDN
     *          The base DN of the domain the attempt was for.
     * @return {@code true} if this is to be reported, {@code false} if nothing was reported
     *         about that peer or if it was already reported as answering without a session.
     */
    boolean recordReachableWithoutSession(final HostPort peer, final DN baseDN)
    {
      final ConcurrentMap<DN, Reported> domains = reported.get(peer);
      return domains != null
          && Reported.DOWN == domains.replace(baseDN, Reported.REACHABLE_NO_SESSION);
    }

    /**
     * Records a session established with the provided peer for the provided domain.
     *
     * @param peer
     *          The address of the replication server which was connected to.
     * @param baseDN
     *          The base DN of the domain the connection is for.
     * @return {@code true} if this session ends something which was reported and is
     *         therefore to be reported as well, {@code false} if nothing was reported about
     *         that peer.
     */
    boolean recordConnected(final HostPort peer, final DN baseDN)
    {
      final ConcurrentMap<DN, Reported> domains = reported.get(peer);
      return domains != null && domains.remove(baseDN) != null;
    }

    /**
     * Forgets what was recorded for the peers and the domains which are no longer
     * configured, and which nothing can therefore report a recovery for.
     *
     * @param peers
     *          The addresses of the replication servers which are configured.
     * @param baseDNs
     *          The base DNs of the domains which exist.
     */
    void retainAll(final Set<HostPort> peers, final Set<DN> baseDNs)
    {
      reported.keySet().retainAll(peers);
      for (ConcurrentMap<DN, Reported> domains : reported.values())
      {
        domains.keySet().retainAll(baseDNs);
      }
    }

    private ConcurrentMap<DN, Reported> domainsOf(final HostPort peer)
    {
      return reported.computeIfAbsent(peer, unused -> new ConcurrentHashMap<>());
    }
  }

  /**

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

@@ -137,11 +137,21 @@
  /**
   * Connect the hosting RS to the RS represented by THIS handler
   * on an outgoing connection.
   * <p>
   * A handshake which does not complete is aborted rather than thrown out of here, and
   * three of its seven aborts carry no message at all: a peer which stops the handshake
   * answers with a {@link StopMsg}, which {@code Session.close()} publishes for every
   * abort of its own, and {@link #abortStart} logs nothing when the reason is null. The
   * caller is what is left to tell a connection from an attempt which only reached the
   * replication port of a peer.
   *
   * @param baseDN The baseDN
   * @param sslEncryption The sslEncryption requested to the remote RS.
   * @return {@code true} when the handshake completed and this handler is started,
   *         {@code false} when it was aborted.
   * @throws DirectoryException when an error occurs.
   */
  public void connect(DN baseDN, boolean sslEncryption)
  public boolean connect(DN baseDN, boolean sslEncryption)
      throws DirectoryException
  {
    // we are the initiator and decides of the encryption
@@ -180,7 +190,7 @@
              .getClass().getCanonicalName(), "ReplServerStartMsg");
          abortStart(message);
        }
        return;
        return false;
      }

      processStartFromRemote((ReplServerStartMsg) msg);
@@ -189,7 +199,7 @@
      {
        // Simultaneous cross connect.
        abortStart(null);
        return;
        return false;
      }

      /*
@@ -230,7 +240,7 @@
        {
          // Simultaneous cross connect.
          abortStart(null);
          return;
          return false;
        }

        logTopoHandshakeSNDandRCV(outTopoMsg, inTopoMsg);
@@ -252,6 +262,7 @@
          replicationServerDomain.getBaseDN(), session.getReadableRemoteAddress());

      super.finalizeStart();
      return true;
    }
    catch (IOException e)
    {
@@ -260,16 +271,19 @@
          getReplicationServerId(),
          session.getReadableRemoteAddress());
      abortStart(errMessage);
      return false;
    }
    catch (DirectoryException e)
    {
      logger.traceException(e);
      abortStart(e.getMessageObject());
      return false;
    }
    catch (Exception e)
    {
      logger.traceException(e);
      abortStart(LocalizableMessage.raw(e.getLocalizedMessage()));
      return false;
    }
    finally
    {

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java

@@ -1161,8 +1161,18 @@
    catch (Exception e)
    {
      logger.traceException(e);
      errorMessage = WARN_EXCEPTION_STARTING_SESSION_PHASE.get(
          getServerId(), serverURL, getBaseDN(), stackTraceToSingleLineString(e));
      /*
       * The whole trace, on one line, for the server this broker is electing, which is
       * what this message carried before it was reported for the others as well; the
       * message alone for a server which was only contacted, whose report is not
       * throttled. An SSLException lands here, is not paced by a connect timeout the way
       * a refused connection is, and collectReplicationServersInfo() reruns over every
       * URL on each reconnection: one bad certificate among several replication servers
       * would otherwise write a full stack trace per flap cycle. The trace of those is in
       * the trace log, which traceException() above wrote it to.
       */
      errorMessage = WARN_EXCEPTION_STARTING_SESSION_PHASE.get(getServerId(), serverURL, getBaseDN(),
          keepSession ? stackTraceToSingleLineString(e) : getExceptionMessage(e));
    }
    finally
    {
@@ -1172,16 +1182,51 @@
        close(socket);
      }

      if (!hasConnected && errorMessage != null && !connectionError)
      if (!hasConnected && errorMessage != null)
      {
        // There was no server waiting on this host:port
        // Log a notice and will try the next replicationServer in the list
        if (keepSession) // Log error message only for final connection
        if (!connectionError)
        {
          // log the error message only once to avoid overflowing the error log
          logger.error(errorMessage);
          /*
           * Report the cause for every replication server contacted, and not only for the
           * elected one: none is ever elected when none of them answers, and this is then
           * the only place naming why -- a refused connection, a rejected certificate, a
           * wrong port -- next to the "unable to connect to any replication servers"
           * summary which names none of them.
           *
           * connectionError is what bounds the volume, and it bounds it to one line per
           * replication server and per attempt to connect this broker makes. It is set
           * when an attempt reaches no replication server at all, and stays set until a
           * session is established, so the 500 ms loop which retries a total outage
           * reports its first pass only.
           *
           * It is not set while this broker is connected, so the unreachable servers of a
           * topology which still serves this broker are reported again on each
           * reconnection -- one line each, so a reconnection costs as many lines as there
           * are servers it could not reach, where it used to cost none. That is the volume
           * this reporting is worth: a broker reconnects when its session is lost, not on
           * a schedule, and a server which cannot be reached over several reconnections is
           * a server whose configuration or certificate needs looking at.
           *
           * The severity says what the failure cost this broker, not what the message is
           * named: the elected server keeps the error it was reported with, and a server
           * which was only contacted is a warning, so that a broker which does find a
           * server to work with does not raise an error over the one it did not need.
           * ERR_DS_DN_DOES_NOT_MATCH is the one message which reaches the second branch
           * under an ERR_ name -- it is set without setting hasConnected -- and it is a
           * permanent misconfiguration rather than a transient. It still goes out as an
           * error for the server this broker is electing, which is the one it cannot work
           * without, and where it is only contacted the broker has another server to work
           * with. Before this, that path logged nothing above trace either way.
           */
          if (keepSession)
          {
            logger.error(errorMessage);
          }
          else
          {
            logger.warn(errorMessage);
          }
        }

        logger.trace(errorMessage);
      }
    }

 opendj-server-legacy/src/main/java/org/opends/server/util/FailureLogThrottle.java

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

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Bounds how often a repeating failure is reported, so that a failure which a retry loop
 * or a stream of unwanted connections reproduces many times a second is still visible in
 * the error log without filling it.
 * <p>
 * One failure is logged per interval and the failures in between are counted, so that the
 * next one logged reports how many it stands for and a single line cannot be mistaken for
 * a single failure. That count looks backwards only: the failures following the last one
 * logged of a burst are counted but never reported, as nothing flushes the count when the
 * failures stop.
 * <p>
 * The first failure is always logged, however long the throttle has existed before it.
 * <p>
 * Safe to record from several threads, and exact for one: a failure recorded while
 * another thread is opening a new interval may be counted in the interval which is
 * closing or in the one which opens, so the count a caller reports can be off by the
 * failures which raced with it. It is a count of failures, not a ledger.
 */
public final class FailureLogThrottle
{
  /** Minimum interval, in nanoseconds, between two failures logged. */
  private final long intervalNanos;

  /**
   * Value of {@link System#nanoTime()} at which the last failure was logged. It starts one
   * interval in the past so that the first failure is logged.
   */
  private final AtomicLong lastLogNanos;

  /** Number of failures suppressed since the last one logged. */
  private final AtomicLong suppressed = new AtomicLong();

  /**
   * Creates a throttle logging at most one failure per provided interval.
   *
   * @param interval
   *          The interval between two failures logged.
   * @param unit
   *          The unit the interval is expressed in.
   */
  public FailureLogThrottle(final long interval, final TimeUnit unit)
  {
    this.intervalNanos = unit.toNanos(interval);
    this.lastLogNanos = new AtomicLong(System.nanoTime() - intervalNanos);
  }

  /**
   * Records a failure which happened at the provided time and tells how it must be logged,
   * together with the number of failures suppressed since the previous one logged.
   *
   * @param nowNanos
   *          The value of {@link System#nanoTime()} at which the failure happened.
   * @return A number greater than or equal to zero if this failure is to be logged, which is
   *         then the number of failures suppressed since the previous one logged, or
   *         {@code -count - 1} if this failure is itself to be suppressed, where
   *         {@code count} is the number of failures suppressed since the previous one
   *         logged, this one included.
   */
  public long record(final long nowNanos)
  {
    final long lastLog = lastLogNanos.get();
    if (nowNanos - lastLog >= intervalNanos && lastLogNanos.compareAndSet(lastLog, nowNanos))
    {
      return suppressed.getAndSet(0);
    }
    return -suppressed.incrementAndGet() - 1;
  }
}

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -447,8 +447,9 @@
 misconfigured: check that the certificate nickname configured in the crypto \
 manager exists in the ads-truststore, and that every server of the topology \
 trusts the certificates of the other servers. At most one such failure is \
 logged as a warning every %d minutes; the others go to the debug log, which is \
 disabled by default (%d since the previous warning). The error was: %s
 logged as a warning every %d minutes; the others are recorded with the \
 information severity, which the replication log publishes and the error log \
 does not (%d since the previous warning). The error was: %s
WARN_MISSING_REMOTE_MONITOR_DATA_106=Timed out waiting for monitor data \
 for the domain "%s" from replication server RS(%d)
NOTE_LOAD_BALANCE_REPLICATION_SERVER_213=Directory Server DS(%d) is disconnecting \
@@ -626,6 +627,36 @@
NOTE_REPLAY_ABANDONED_CHANGE_309=Could not replay change %s in domain "%s": the replay thread \
 it was given to is stopping. The change has not been recorded as replayed and is given back to \
 the replication server, which still owns it
WARN_REPLICATION_SERVER_ACCEPT_ERROR_310=Replication server RS(%d) could not accept a connection on %s: %s. \
 A failure which repeats at once, the process running out of file descriptors for instance, makes the \
 listen thread wait before accepting again, so that it does not spin. At most one such failure is logged \
 as a warning every %d minutes; the others are recorded with the information severity, which the \
 replication log publishes and the error log does not (%d since the previous warning)
WARN_REPLICATION_SERVER_SESSION_SETUP_ERROR_311=Replication server RS(%d) accepted a connection from %s \
 but could not start a replication session on it, and closed it: %s. A failed SSL handshake is reported \
 on its own, so this is something else. It may be benign, a network probe which connects and closes \
 without saying anything, but it also occurs when the ads-truststore cannot be read, which fails every \
 inbound connection this way: where the connection came from a replication peer, check that the \
 ads-truststore still exists and is readable by the server. At most one such failure is logged as a \
 warning every %d minutes; the others are recorded with the information severity, which the replication \
 log publishes and the error log does not (%d since the previous warning)
WARN_REPLICATION_SERVER_CONNECT_ERROR_312=Replication server RS(%d) could not connect to replication \
 server %s for domain "%s": %s. This is reported once: the next message about this replication server \
 comes when it can be reached again, either because this server reached it or because it connected to \
 this server from the address it is configured under. The domain named is the one this attempt was \
 for, and a replication server which cannot be reached is not tried for the other domains it serves \
 while it cannot be, so this names one of them rather than all of them
NOTE_REPLICATION_SERVER_CONNECT_RESTORED_313=Replication server RS(%d) connected to replication server \
 %s for domain "%s", which it had reported it had no replication session with
WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION_314=Replication server RS(%d) reached replication server \
 %s for domain "%s", which it had reported it could not connect to, but the handshake with it did not \
 complete: this attempt established no replication session. Where this server ended the handshake, the \
 reason is logged next to this message, whether it rejected the peer or the peer went away while the \
 handshake was running. Where nothing is logged next to it, the peer ended the handshake, and it may \
 have logged no reason of its own either: a shutdown under way, or a connection the two servers made \
 to each other at the same time, of which one is dropped while the other one serves the domain. Until \
 a handshake completes, no change is replicated over this connection, and the connection which \
 completes one is reported in its turn
WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \
 the replay of one of its changes to finish. A change which reaches the backend from now on \
 is not recorded in the ServerState being saved, so the replication server sends it again \

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

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

import static org.assertj.core.api.Assertions.assertThat;
import static org.opends.messages.ReplicationMessages.*;

import java.util.List;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.DomainFakeCfg;
import org.opends.server.replication.plugin.DummyReplicationDomain;
import org.opends.server.replication.service.ReplicationBroker;
import org.testng.annotations.Test;

/**
 * Tests what a directory server logs when it cannot connect to any replication server.
 */
@SuppressWarnings("javadoc")
public class ReplicationBrokerConnectFailureTest extends ReplicationTestCase
{
  /**
   * Tests that a directory server which reaches no replication server at all names the
   * reason it reached none.
   * <p>
   * The cause is built for every replication server contacted, but it used to be logged
   * only for the elected one, and no server is ever elected when none of them answers: a
   * rejected certificate, a refused connection and a wrong port then all read as "unable
   * to connect to any replication servers".
   */
  @Test
  public void aBrokerWhichReachesNoReplicationServerNamesTheCause() throws Exception
  {
    TestCaseUtils.startServer();

    final DN baseDN = DN.valueOf(TestCaseUtils.TEST_ROOT_DN_STRING);
    final int serverId = 4021;
    // Free, and left free: nothing must listen on it for this test to be about a failure.
    final int deadPort = TestCaseUtils.findFreePorts(1)[0];
    final String deadServer = "127.0.0.1:" + deadPort;

    final DomainFakeCfg config = newFakeCfg(baseDN, serverId, deadPort);
    final ReplicationBroker broker = new ReplicationBroker(
        new DummyReplicationDomain(0), new ServerState(), config, getReplSessionSecurity());
    try
    {
      final List<String> records = errorLogRecordsOf(() -> {
        broker.start();
        return null;
      });

      assertThat(broker.isConnected()).as("nothing listens on " + deadServer).isFalse();
      final String logged = records.toString();
      assertThat(logged)
          .as("the directory server should name why it could not connect to " + deadServer)
          .contains(WARN_NO_CHANGELOG_SERVER_LISTENING.get(serverId, deadServer, baseDN).toString());
      assertThat(logged)
          .as("the summary of the outage should still be logged next to the cause")
          .contains(WARN_NO_AVAILABLE_CHANGELOGS.get(serverId, baseDN).toString());
      /*
       * A replication server which was only contacted is reported as a warning, and the
       * one this broker elects keeps the error it was reported with. No server answers
       * here, so none is elected and every record about one is a warning: the severity is
       * part of what this reports, not a detail of how, and asserting the text alone would
       * leave the split which decides it untested.
       */
      assertThat(recordOf(records, WARN_NO_CHANGELOG_SERVER_LISTENING.get(serverId, deadServer, baseDN)))
          .as("the cause of a replication server which was only contacted is a warning")
          .contains("severity=WARNING");
    }
    finally
    {
      stop(broker);
    }
  }
}

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

@@ -32,6 +32,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -57,6 +58,10 @@
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ModifyOperation;
import org.opends.server.loggers.ErrorLogPublisher;
import org.opends.server.loggers.ErrorLogger;
import org.opends.server.loggers.TextErrorLogPublisher;
import org.opends.server.loggers.TextWriter;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.SearchRequest;
@@ -939,6 +944,227 @@
    return new ReplSessionSecurity(null, null, null, true);
  }

  /**
   * Runs the provided action and returns the records the error log received while it ran.
   * <p>
   * A publisher of its own is registered for the duration rather than reading the one the
   * test harness installs, whose contents span the whole test JVM.
   * <p>
   * It publishes every severity, so that a record a throttle kept out of the warnings is
   * captured too. That is server wide while the action runs, so the records of other
   * threads are captured as well and a caller has to pick out its own.
   *
   * @param action
   *          The action to run.
   * @return The error log records written while the action ran, in order.
   * @throws Exception
   *           Whatever the action throws.
   */
  protected static List<String> errorLogRecordsOf(final Callable<Void> action) throws Exception
  {
    return errorLogRecordsOf(unused -> action.call());
  }

  /**
   * Runs the provided action, which reads the error log records as they are written, and
   * returns the records the error log received while it ran.
   * <p>
   * The list handed to the action is the live one, so an action can wait for a server
   * thread to log something instead of waiting for a duration. It is synchronized, and
   * iterating it is not: a reader has to copy it, or hold its monitor.
   *
   * @param action
   *          The action to run, taking the records written so far.
   * @return The error log records written while the action ran, in order.
   * @throws Exception
   *           Whatever the action throws.
   */
  // The publisher is built raw and handed to a parameterized addLogPublisher: the
  // conversion is unchecked, and it is the one the test harness makes as well.
  @SuppressWarnings({ "rawtypes", "unchecked" })
  protected static List<String> errorLogRecordsOf(ErrorLogAction action) throws Exception
  {
    final List<String> records = Collections.synchronizedList(new ArrayList<String>());
    final ErrorLogPublisher capture =
        TextErrorLogPublisher.getToolStartupTextErrorPublisher(new TextWriter()
        {
          @Override
          public void writeRecord(String record)
          {
            records.add(record);
          }

          @Override
          public void flush()
          {
            // Nothing is buffered.
          }

          @Override
          public void shutdown()
          {
            // Nothing is buffered.
          }

          @Override
          public long getBytesWritten()
          {
            return 0;
          }
        });
    ErrorLogger.getInstance().addLogPublisher(capture);
    try
    {
      action.run(records);
    }
    finally
    {
      ErrorLogger.getInstance().removeLogPublisher(capture);
    }
    // Copied under the monitor: the publishers are iterated from a snapshot, so a thread
    // which is already inside it can still write a record after this one was removed, and
    // the caller must not have to synchronize to read what it got.
    return copyOf(records);
  }

  /**
   * Returns the one record of the provided error log which reports the provided message.
   * <p>
   * A record holds the severity it was published with next to the message, so this is how
   * a test reads the severity of a message it expects, rather than only its text.
   *
   * @param records
   *          The error log records to look in, as {@link #errorLogRecordsOf} returned them.
   * @param message
   *          The message the record is expected to report.
   * @return The record reporting the provided message.
   */
  protected static String recordOf(List<String> records, LocalizableMessage message)
  {
    String found = null;
    for (String record : records)
    {
      if (record.contains(message.toString()))
      {
        assertNull(found, "\"" + message + "\" should have been logged once, but the error log"
            + " holds it more than once: " + records);
        found = record;
      }
    }
    assertNotNull(found, "\"" + message + "\" should have been logged, but the error log holds: " + records);
    return found;
  }

  /**
   * Returns how many records of the provided error log hold the provided text.
   *
   * @param records
   *          The error log records to count in.
   * @param contained
   *          The text the counted records hold.
   * @return The number of records holding the provided text.
   */
  protected static int countRecordsOf(List<String> records, String contained)
  {
    int count = 0;
    synchronized (records)
    {
      for (String record : records)
      {
        if (record.contains(contained))
        {
          count++;
        }
      }
    }
    return count;
  }

  /**
   * Waits for a record holding the provided text to be written to the provided error log.
   * <p>
   * For what a server thread logs on its own schedule: the connect thread of a replication
   * server retries a peer every second, so what it reports is waited for rather than
   * expected to be there already.
   * <p>
   * Only the records which arrived since the previous poll are read: the capture publishes
   * every severity, which is every {@code logger.debug} of every thread of the server for
   * as long as it is installed, so rereading the whole list on each poll would cost the
   * square of what a slow wait captures.
   *
   * @param records
   *          The live error log records, as {@link ErrorLogAction} received them.
   * @param contained
   *          The text the awaited record holds.
   * @param timeoutMs
   *          How long to wait for it, in milliseconds.
   * @throws InterruptedException
   *           If the wait is interrupted.
   */
  protected static void waitForErrorLogRecord(List<String> records, String contained, long timeoutMs)
      throws InterruptedException
  {
    final long deadline = System.currentTimeMillis() + timeoutMs;
    int read = 0;
    while (true)
    {
      synchronized (records)
      {
        final int written = records.size();
        while (read < written)
        {
          if (records.get(read++).contains(contained))
          {
            return;
          }
        }
      }
      // Tested after the records are read and not after the sleep, so that what arrives
      // during the last sleep of the wait is still read: it arrived inside the timeout.
      if (System.currentTimeMillis() >= deadline)
      {
        break;
      }
      Thread.sleep(100);
    }
    fail("\"" + contained + "\" should have been logged within " + timeoutMs
        + " ms, but the error log received " + read + " records, the last of them: "
        + lastRecordsOf(records));
  }

  /** Copies the live error log records, which are synchronized but not safe to iterate. */
  private static List<String> copyOf(List<String> records)
  {
    synchronized (records)
    {
      return new ArrayList<>(records);
    }
  }

  /**
   * The tail of the live error log records, for a failure message: the capture holds every
   * severity, so the whole list is not something a report can carry.
   */
  private static List<String> lastRecordsOf(List<String> records)
  {
    final List<String> all = copyOf(records);
    return all.subList(Math.max(0, all.size() - 20), all.size());
  }

  /** An action which reads the error log records as they are written. */
  protected interface ErrorLogAction
  {
    /**
     * Runs the action.
     *
     * @param records
     *          The live error log records written since the action started.
     * @throws Exception
     *           Whatever the action throws.
     */
    void run(List<String> records) throws Exception;
  }

  protected void executeTask(Entry taskEntry, long maxWaitTimeInMillis) throws Exception
  {
    addTask(taskEntry, ResultCode.SUCCESS, null);

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

New file
@@ -0,0 +1,234 @@
/*
 * 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.opends.server.util.CollectionUtils.newHashSet;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.server.ReplicationServer.ConnectFailureReporter;
import org.opends.server.types.HostPort;
import org.testng.annotations.Test;

/** Tests for {@link ConnectFailureReporter}. */
@SuppressWarnings("javadoc")
public class ConnectFailureReporterTest extends DirectoryServerTestCase
{
  /*
   * Peers which differ by their port: HostPort resolves a host which is not "localhost"
   * through InetAddress.getByName and compares what it resolved, so two host names would
   * be one peer on a resolver which answers both, and a lookup each on one which answers
   * neither.
   */
  private static final HostPort PEER = HostPort.valueOf("localhost:8989");
  private static final HostPort OTHER_PEER = HostPort.valueOf("localhost:8990");
  private static final DN DOMAIN = DN.valueOf("dc=example,dc=com");
  private static final DN OTHER_DOMAIN = DN.valueOf("dc=other,dc=com");

  @Test
  public void theFirstFailureIsReported() throws Exception
  {
    assertThat(new ConnectFailureReporter().recordFailure(PEER, DOMAIN))
        .as("the first failure to connect to a peer is reported").isTrue();
  }

  @Test
  public void aPeerWhichStaysUnreachableIsReportedOnce() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);

    assertThat(reporter.recordFailure(PEER, DOMAIN))
        .as("the connect thread retries every few seconds, and reporting each retry"
            + " would fill the error log for as long as the peer is down").isFalse();
  }

  @Test
  public void eachPeerAndDomainIsReportedOnItsOwn() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);

    assertThat(reporter.recordFailure(OTHER_PEER, DOMAIN))
        .as("another peer is another failure").isTrue();
    assertThat(reporter.recordFailure(PEER, OTHER_DOMAIN))
        .as("the same peer for another domain is another failure").isTrue();
  }

  @Test
  public void aConnectionEstablishedAfterAReportedFailureIsReported() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);

    assertThat(reporter.recordConnected(PEER, DOMAIN))
        .as("the failure was reported, so the recovery has to be reported as well").isTrue();
  }

  @Test
  public void aConnectionEstablishedWithoutAReportedFailureIsNotReported() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    assertThat(reporter.recordConnected(PEER, DOMAIN))
        .as("every connection the topology establishes would otherwise be reported").isFalse();
    reporter.recordFailure(OTHER_PEER, DOMAIN);
    assertThat(reporter.recordConnected(PEER, DOMAIN))
        .as("the failure of another peer is not this peer's").isFalse();
  }

  @Test
  public void aPeerWhichAnswersWithoutASessionAfterAnOutageIsReported() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);

    assertThat(reporter.recordReachableWithoutSession(PEER, DOMAIN))
        .as("a peer which was reported unreachable and now answers is no longer unreachable,"
            + " and what it is instead is not what the outage said").isTrue();
  }

  @Test
  public void aPeerWhichAnswersWithoutASessionWithoutAnOutageIsNotReported() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    assertThat(reporter.recordReachableWithoutSession(PEER, DOMAIN))
        .as("nothing was reported about this peer, so there is no outage to close: an abort"
            + " which reports itself is the handshake's to report").isFalse();
  }

  @Test
  public void aPeerWhichKeepsAnsweringWithoutASessionIsReportedOnce() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordReachableWithoutSession(PEER, DOMAIN);

    assertThat(reporter.recordReachableWithoutSession(PEER, DOMAIN))
        .as("the connect thread retries every few seconds, and a peer which aborts every"
            + " handshake would otherwise be reported on each of them").isFalse();
  }

  /**
   * The recovery which follows an answer without a session is the one a present-or-absent
   * record cannot report: the answer would have consumed the record, and the handshake which
   * completes seconds later would find nothing left to close. A peer restarting takes that
   * path -- its port answers before its domains are up -- and the operator would be left with
   * a warning saying no change is replicated over a connection which is replicating.
   */
  @Test
  public void aSessionEstablishedAfterAnAnswerWithoutOneIsReported() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordReachableWithoutSession(PEER, DOMAIN);

    assertThat(reporter.recordConnected(PEER, DOMAIN))
        .as("the last thing reported about this peer said it had no session, so the session"
            + " it now has has to be reported").isTrue();
  }

  @Test
  public void aPeerWhichGoesDownAfterAnsweringWithoutASessionIsReportedAgain() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordReachableWithoutSession(PEER, DOMAIN);

    assertThat(reporter.recordFailure(PEER, DOMAIN))
        .as("a peer which answered and now does not is unreachable again, which is not what"
            + " the answer without a session reported").isTrue();
  }

  @Test
  public void aPeerWhichLeavesTheConfigurationIsForgotten() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordFailure(OTHER_PEER, DOMAIN);

    reporter.retainAll(newHashSet(PEER), newHashSet(DOMAIN));

    assertThat(reporter.recordFailure(OTHER_PEER, DOMAIN))
        .as("nothing connects to a peer which is no longer configured, so nothing would"
            + " ever clear what was recorded for it: its next failure has to be reported").isTrue();
    assertThat(reporter.recordFailure(PEER, DOMAIN))
        .as("a peer which is still configured keeps what was recorded for it").isFalse();
  }

  @Test
  public void aDomainWhichIsRemovedIsForgotten() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordFailure(PEER, OTHER_DOMAIN);

    reporter.retainAll(newHashSet(PEER), newHashSet(DOMAIN));

    assertThat(reporter.recordFailure(PEER, OTHER_DOMAIN))
        .as("the domain is gone, so the failure recorded for it can no longer be cleared").isTrue();
    assertThat(reporter.recordFailure(PEER, DOMAIN))
        .as("the domain which remains keeps what was recorded for it").isFalse();
  }

  @Test
  public void aPeerFailingAgainAfterAConnectionIsReportedAgain() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordConnected(PEER, DOMAIN);

    assertThat(reporter.recordFailure(PEER, DOMAIN))
        .as("a peer which goes down again is a new failure, not the one already reported").isTrue();
  }

  /**
   * A session is established for one domain at a time, and what a connection ends is the
   * outage of the domain it was made for. Forgetting the peer instead leaves the domains
   * which are still down looking like domains nothing was reported about: each of them is
   * reported down a second time on the next pass of the connect thread, and reported
   * recovered on the pass after that, for as long as they stay down.
   */
  @Test
  public void aConnectionForOneDomainLeavesTheOtherDomainsOfThePeerRecorded() throws Exception
  {
    final ConnectFailureReporter reporter = new ConnectFailureReporter();

    reporter.recordFailure(PEER, DOMAIN);
    reporter.recordFailure(PEER, OTHER_DOMAIN);

    reporter.recordConnected(PEER, DOMAIN);

    assertThat(reporter.recordFailure(PEER, OTHER_DOMAIN))
        .as("the other domain of that peer is still down, and its failure is the one already"
            + " reported rather than a new one").isFalse();
    assertThat(reporter.recordConnected(PEER, OTHER_DOMAIN))
        .as("the outage reported for the other domain is still open, so the connection which"
            + " ends it is still to be reported").isTrue();
  }
}

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

New file
@@ -0,0 +1,817 @@
/*
 * 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.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.util.CollectionUtils.newArrayList;
import static org.opends.server.util.CollectionUtils.newTreeSet;
import static org.opends.server.util.StaticUtils.*;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StopMsg;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.types.HostPort;
import org.testng.annotations.Test;

/**
 * Tests what a replication server logs about the replication servers it connects to,
 * through the replication server rather than through the bookkeeping alone.
 * <p>
 * What is reported is decided in two places -- {@code connect()} and the already connected
 * branch of {@code runConnect()} -- and the decision each of them makes is only correct
 * with respect to the other: a test of {@link ReplicationServer.ConnectFailureReporter} on
 * its own passes with either of them deleted, or with any of the conditions an outage used
 * to be closed on put back in front of it.
 * <p>
 * The tests which drive a peer of their own drive one the already connected branch of
 * {@code runConnect()} cannot report anything about, so that {@code connect()} is the only
 * place a recovery can come from: a peer registered with no domain at all, and one
 * registered under an address which is not the one it is configured under. Both shapes end
 * on a session {@code abortStart} closed and on a peer which is connected for no domain,
 * which is what makes them the outages this server has to close without reading either --
 * and, being aborts, the recoveries it must not report as connections.
 * <p>
 * The one case whose peer does end up registered is
 * {@link #aSessionEstablishedAfterAnAnswerWithoutOneIsReported}, where the session is the
 * point: what it pins is that the answer without a session left something for the session
 * to close, which holds wherever the session is reported from.
 * <p>
 * Every test configures the peer it drives, and creates the domain it drives it for, before
 * anything is reported. The connect thread runs {@code retainAll} against the configured
 * peers and the domains of the pass on every pass, so a peer or a domain it does not see
 * has whatever was recorded for it cleared about once a second -- which is a record the
 * test is not allowed to rely on. Where a record is made from the test thread rather than
 * by the connect thread, one whole pass is waited for as well, so that no pass which
 * started before the domain existed is still holding a snapshot without it.
 */
@SuppressWarnings("javadoc")
public class ReplicationServerConnectFailureTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  /** How long a peer which is down, or which came back, is waited to be reported within. */
  private static final long REPORT_TIMEOUT_MS = 30000;

  private static final int RS_ID = 8241;
  private static final int PEER_RS_ID = 8242;

  /**
   * Tests that a replication server reports a peer it cannot connect to once, and reports
   * the connection which ends that outage.
   * <p>
   * Both servers run their own connect thread, so this goes through {@code runConnect()}:
   * the peer is reported by the connect thread of the server under test, and the
   * connection which ends the outage is reported by whichever of {@code connect()} and the
   * already connected branch wins -- the peer dials back, and either end may connect first.
   */
  @Test
  public void aPeerWhichIsDownIsReportedOnceAndItsReturnIsReported() throws Exception
  {
    TestCaseUtils.startServer();

    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final int[] ports = TestCaseUtils.findFreePorts(2);
    final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
    final String restored = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();

    // Held rather than returned: whatever the capture below throws, these are what the
    // listen ports, the threads and the changelogs of both servers hang on.
    final ReplicationServer[] servers = new ReplicationServer[2];
    try
    {
      final List<String> records = errorLogRecordsOf(live -> {
        /*
         * Built inside the capture: the constructor starts the connect thread, and the
         * failure to reach a peer which is down is reported once for the whole outage, so
         * a report which precedes the capture is a report which never arrives.
         */
        servers[0] = new ReplicationServer(new ReplServerFakeConfiguration(
            ports[0], "replicationServerConnectFailureDb", 0, RS_ID, 0, 100,
            newTreeSet(peerAddress.toString())));
        // The connect thread iterates the domains of this server, so it has nothing to
        // connect for, and nothing to keep a record for, until there is one.
        servers[0].getReplicationServerDomain(baseDN, true);

        waitForErrorLogRecord(live, connectErrorPrefix(peerAddress, baseDN), REPORT_TIMEOUT_MS);

        servers[1] = new ReplicationServer(new ReplServerFakeConfiguration(
            ports[1], "replicationServerConnectFailurePeerDb", 0, PEER_RS_ID, 0, 100,
            newTreeSet(HostPort.localAddress(ports[0]).toString())));
        servers[1].getReplicationServerDomain(baseDN, true);

        waitForErrorLogRecord(live, restored, REPORT_TIMEOUT_MS);
      });

      /*
       * The connect thread retries a peer which is down every second, and it stayed down
       * for as long as the wait above took: one line for the outage, one for its end.
       */
      assertThat(countRecordsOf(records, connectErrorPrefix(peerAddress, baseDN)))
          .as("a peer which stays down should be reported once, not on every retry")
          .isEqualTo(1);
      assertThat(countRecordsOf(records, restored))
          .as("the end of the outage should be reported once")
          .isEqualTo(1);
    }
    finally
    {
      remove(servers[1], servers[0]);
    }
  }

  /**
   * Tests that a peer which stops the handshake it is offered closes the outage reported
   * for it, and is reported reachable without a session rather than connected.
   * <p>
   * The outage is a failure to connect, so what closes it is the peer answering:
   * {@code WARN_REPLICATION_SERVER_CONNECT_ERROR} is reported for the socket and for the
   * session built on it, the handshake throwing nothing of its own. Holding the outage open
   * across an abort silences the peer this server never sees connected under the address it
   * dialled -- the multi homed peer of
   * {@link #aPeerRegisteredUnderAnotherAddressStillClosesItsOutage}, and one protocol
   * version down a peer which negotiates V1, connected and never registered.
   * <p>
   * Reporting it connected instead is the other half of the same line.
   * {@code Session.close()} publishes a {@code StopMsg} for every abort its own end makes
   * and {@code abortStart(null)} logs nothing, so a peer which rejects this server -- its
   * own duplicate server id, a cross connect it resolves against this server, a shutdown
   * under way -- would have "connected" as the last thing this server ever says about it.
   */
  @Test
  public void aPeerWhichStopsTheHandshakeStillClosesItsOutage() throws Exception
  {
    aPeerWhichAnswersClosesItsOutage("replicationServerHandshakeAbortDb",
        (session, received) -> {
          session.publish(new StopMsg());
          // Read until the peer has read the StopMsg and closed its own end: closing this
          // one first would race that read, and turn the abort into a failed handshake.
          session.receive();
        });
  }

  /**
   * Tests that the session a peer establishes after it answered without one is reported.
   * <p>
   * This is the peer restarting, which is the ordinary way into the middle state: its
   * replication port answers before its domains are up, so it stops the handshake, and the
   * handshake which completes seconds later is what an operator is waiting for. The only
   * other line that session writes is
   * {@code logger.debug(INFO_REPLICATION_SERVER_CONNECTION_TO_RS)}, which is the
   * {@code information} severity: the replication log holds it and the error log does not,
   * so a recovery not reported here is a warning left standing over a connection which
   * replicates.
   * <p>
   * What that needs is a record which holds the last state <em>reported</em> rather than
   * whether an outage is open, and it is the call site rather than the bookkeeping which
   * has to hold it: reporting the answer without a session through
   * {@code recordConnected()} passes every test of
   * {@link ReplicationServer.ConnectFailureReporter} and consumes the record all the same.
   * <p>
   * The peer answers on one socket throughout, and the outage is never reopened in
   * between: a peer which stops answering is a new outage, and the session would then close
   * that one instead of the answer without a session, which is what this case is about.
   */
  @Test
  public void aSessionEstablishedAfterAnAnswerWithoutOneIsReported() throws Exception
  {
    TestCaseUtils.startServer();

    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final int[] ports = TestCaseUtils.findFreePorts(2);
    final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
    final String reachable =
        WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
    final String connected = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();

    final ReplicationServer[] servers = new ReplicationServer[1];
    final ExecutorService peerThread = Executors.newSingleThreadExecutor();
    try
    {
      final ReplSessionSecurity security = getReplSessionSecurity();
      // What the peer answers with, flipped once the answer without a session has been
      // reported: the domains of a peer whose port answers come up while it is answering.
      final AtomicBoolean domainsAreUp = new AtomicBoolean();
      final List<String> records = errorLogRecordsOf(live -> {
        servers[0] = new ReplicationServer(new ReplServerFakeConfiguration(
            ports[0], "replicationServerSessionAfterAbortDb", 0, RS_ID, 0, 100,
            newTreeSet(peerAddress.toString())));
        final ReplicationServer rs = servers[0];
        rs.getReplicationServerDomain(baseDN, true);
        // The pass a record made from this thread has to outlive, as in
        // aPeerWhichAnswersClosesItsOutage.
        rs.waitConnections();

        assertThat(rs.connect(peerAddress, baseDN))
            .as("nothing listens on " + peerAddress).isFalse();
        waitForErrorLogRecord(live, connectErrorPrefix(peerAddress, baseDN), REPORT_TIMEOUT_MS);

        final AtomicBoolean serving = new AtomicBoolean(true);
        try (ServerSocket peerSocket = bindPeerPort(ports[1]))
        {
          peerSocket.setSoTimeout(SOCKET_TIMEOUT_MS);
          peerThread.submit(() -> answerEveryHandshake(peerSocket, security, serving,
              (session, received) -> {
                if (!domainsAreUp.get())
                {
                  session.publish(new StopMsg());
                  // Read until the peer has read the StopMsg and closed its own end, as in
                  // aPeerWhichStopsTheHandshakeStillClosesItsOutage.
                  session.receive();
                  return;
                }
                answerWholeHandshake(session, received, peerAddress, baseDN);
              }));

          assertThat(rs.connect(peerAddress, baseDN))
              .as("the peer answers and stops the handshake, so this is no session").isFalse();
          waitForErrorLogRecord(live, reachable, REPORT_TIMEOUT_MS);

          domainsAreUp.set(true);
          /*
           * The return is not asserted: the connect thread dials the same peer, and a
           * session it establishes first leaves this handshake aborting on the handler that
           * session registered. Either of them is the session which ends the middle state,
           * and the record below is what has to report it.
           */
          rs.connect(peerAddress, baseDN);
          waitForErrorLogRecord(live, connected, REPORT_TIMEOUT_MS);
        }
        finally
        {
          serving.set(false);
        }
      });

      assertThat(countRecordsOf(records, connected))
          .as("the session established after the answer without one has to be reported, or the"
              + " warning about a connection which replicates nothing stays the last word about"
              + " a connection which replicates")
          .isEqualTo(1);
      assertThat(countRecordsOf(records, reachable))
          .as("the answer without a session is reported once, however many handshakes the peer"
              + " stops before its domains are up")
          .isEqualTo(1);
      /*
       * What the count above is not enough for: a failure to reach the peer between the two
       * would open an outage of its own, and the session would then be reported for closing
       * that one rather than for closing the answer without a session -- which is the state
       * a record that is merely present or absent cannot hold, and the whole of this case.
       * The peer answers on one socket throughout, so the only failure which can land here
       * is a dial of the connect thread queued behind the handshake below.
       */
      final int reachableAt = indexOfRecord(records, reachable);
      final int connectedAt = indexOfRecord(records, connected);
      assertThat(connectedAt)
          .as("the session is established after the answer without one, not before it")
          .isGreaterThan(reachableAt);
      assertThat(countRecordsOf(records.subList(reachableAt, connectedAt),
              connectErrorPrefix(peerAddress, baseDN)))
          .as("the outage was closed by the answer and not reopened before the session, so the"
              + " session is reported for the answer it follows")
          .isEqualTo(0);
    }
    finally
    {
      peerThread.shutdownNow();
      remove(servers[0]);
    }
  }

  /**
   * Tests that a peer already registered under an address other than the one it is
   * configured under still closes the outage reported for it.
   * <p>
   * This is the multi homed peer, and the reason the recovery can be read neither from the
   * address nor from the session. {@code ServerHandler.toServerAddressURL()} takes the host
   * of a handler from {@code session.getRemoteAddress()} and its port from the start message
   * that handler received, so a peer which dials this server from an address it is not
   * configured under is registered under that other address. Two things follow, and this
   * test drives both: the already connected branch of {@code runConnect()} compares the
   * configured address against one which never matches it, so it can close nothing; and the
   * handshake this server offers that same peer runs into a handler holding its server id
   * under another address URL, which is {@code ERR_DUPLICATE_REPLICATION_SERVER_ID}, an
   * abort of this server rather than of the peer, and a session closed at the end of
   * {@code connect()} for as long as the peer stays where it is.
   * <p>
   * Gating the recovery on either leaves the record of such a peer uncleared for good, and
   * {@code recordFailure()} returns false from then on: the next real outage of it -- the
   * second one counted below -- is not reported at all.
   */
  @Test
  public void aPeerRegisteredUnderAnotherAddressStillClosesItsOutage() throws Exception
  {
    TestCaseUtils.startServer();

    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    /*
     * Three ports: the server under test, the address the peer is configured under and
     * answers on, and the address it registers itself under. The last is never bound --
     * what a multi homed peer costs is that the two addresses are not compared equal, and
     * a port nothing listens on is that, without a second address to bind.
     */
    final int[] ports = TestCaseUtils.findFreePorts(3);
    final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
    final HostPort registeredAs = HostPort.valueOf("127.0.0.1:" + ports[2]);
    final String reachable =
        WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
    final String connected =
        NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();

    final ReplicationServer[] servers = new ReplicationServer[1];
    // Held rather than returned: the registration lasts as long as the session does, and
    // whatever the capture below throws, this is what closes it.
    final Session[] inbound = new Session[1];
    final ExecutorService peerThread = Executors.newSingleThreadExecutor();
    try
    {
      final ReplSessionSecurity security = getReplSessionSecurity();
      final List<String> records = errorLogRecordsOf(live -> {
        servers[0] = new ReplicationServer(new ReplServerFakeConfiguration(
            ports[0], "replicationServerRegisteredAddressDb", 0, RS_ID, 0, 100,
            newTreeSet(peerAddress.toString())));
        final ReplicationServer rs = servers[0];
        rs.getReplicationServerDomain(baseDN, true);
        // The pass a record made from this thread has to outlive, as in
        // aPeerWhichAnswersClosesItsOutage.
        rs.waitConnections();

        assertThat(rs.connect(peerAddress, baseDN))
            .as("nothing listens on " + peerAddress).isFalse();
        waitForErrorLogRecord(live, connectErrorPrefix(peerAddress, baseDN), REPORT_TIMEOUT_MS);

        /*
         * The peer dials this server and completes the handshake from its own side, which
         * is what registers it -- under the address its start message names, not the one it
         * is configured under. Registered before it answers on the configured address, so
         * that every handshake this server offers it from then on, the ones its connect
         * thread offers included, runs into that registration: an attempt which got in
         * before it would abort on the handshake instead, which is the other test.
         */
        inbound[0] = registerPeerFrom(security, ports[0], registeredAs, baseDN);
        /*
         * Waited for rather than assumed: the server registers the handler after it has
         * sent the topology message the handshake above reads, so the registration lands
         * just behind this thread -- and a handshake which ends any other way registers
         * nothing at all, which would leave this case asserting the abort of the other one.
         */
        waitForRegistrationUnder(rs, baseDN, registeredAs, REPORT_TIMEOUT_MS);

        final AtomicBoolean serving = new AtomicBoolean(true);
        try (ServerSocket peerSocket = bindPeerPort(ports[1]))
        {
          peerSocket.setSoTimeout(SOCKET_TIMEOUT_MS);
          // The peer answers every handshake with a well formed start message naming the
          // address it answers on: what aborts the handshake is the handler it already has
          // with this server, not what it answers here.
          peerThread.submit(() -> answerEveryHandshake(peerSocket, security, serving,
              (session, received) -> {
                session.publish(peerStartMsg(peerAddress, baseDN));
                stopEncryptionWith(session, received);
                session.receive();
              }));

          assertThat(rs.connect(peerAddress, baseDN))
              .as("the domain holds a handler for this server id under another address URL,"
                  + " so the handshake aborts on a duplicate server id instead of connecting")
              .isFalse();
          waitForErrorLogRecord(live, reachable, REPORT_TIMEOUT_MS);
        }
        finally
        {
          serving.set(false);
        }

        // And is gone again, which is a new outage: the previous one was closed.
        assertThat(rs.connect(peerAddress, baseDN))
            .as("nothing listens on " + peerAddress).isFalse();
      });

      assertThat(countRecordsOf(records, reachable))
          .as("the peer answered on the address it is configured under, which ends the outage"
              + " reported for it however it is registered with this server")
          .isEqualTo(1);
      assertThat(countRecordsOf(records, connected))
          .as("the handshake aborted on a duplicate server id, so this server has no session"
              + " with the peer over the address it dialled")
          .isEqualTo(0);
      assertThat(countRecordsOf(records, connectErrorPrefix(peerAddress, baseDN)))
          .as("the outage before the peer answered and the one after it went away again are"
              + " two outages, and the second is only reported because the first was closed:"
              + " a record left behind for a peer registered elsewhere silences it for good")
          .isEqualTo(2);
      /*
       * What makes this the multi homing case rather than the one next door: the three
       * counts above are those of a peer which stops the handshake as well, message 314
       * being reported for every abort. Only the registration under another address URL
       * reaches the duplicate server id, and only this server logs it -- the second phase
       * the other case ends on aborts with no message at all.
       */
      assertThat(countRecordsOf(records, ERR_DUPLICATE_REPLICATION_SERVER_ID.get(
              servers[0].getMonitorInstanceName(), registeredAs, peerAddress, PEER_RS_ID).toString()))
          .as("the outbound handshake should have aborted on the duplicate server id of the"
              + " handler registered under another address, not on an unanswered phase two")
          // Not an exact count: the connect thread dials the same peer while it is answering,
          // and every handshake that peer is offered aborts the same way. What this separates
          // is the path, which is one record against none.
          .isGreaterThanOrEqualTo(1);
    }
    finally
    {
      close(inbound[0]);
      peerThread.shutdownNow();
      remove(servers[0]);
    }
  }

  /**
   * Connects to the replication port of the server under test and completes, from the side
   * of a peer replication server, the handshake which registers that peer with the domain.
   * <p>
   * The registration is what this is for, and the handshake has to reach its end to get it:
   * {@code ReplicationServerHandler.startFromRemoteRS()} registers the handler after the
   * second phase, so the topology message of that phase has to be sent and read back.
   *
   * @param security
   *          The session security to build the client session with.
   * @param port
   *          The replication port of the server under test.
   * @param registeredAs
   *          The address this peer names in its start message, and is therefore registered
   *          under, whatever address it dialled from.
   * @param baseDN
   *          The base DN of the domain to register with.
   * @return The session the registration hangs on, closed by the caller.
   */
  private Session registerPeerFrom(ReplSessionSecurity security, int port, HostPort registeredAs, DN baseDN)
      throws Exception
  {
    final Socket socket = new Socket();
    Session session = null;
    try
    {
      socket.setTcpNoDelay(true);
      socket.connect(new InetSocketAddress("127.0.0.1", port), SOCKET_TIMEOUT_MS);
      session = security.createClientSession(socket, SOCKET_TIMEOUT_MS);
      session.publish(peerStartMsg(registeredAs, baseDN));
      session.receive();
      // The initiator of a session decides whether it is encrypted, and the start message
      // above asked for it not to be: both ends leave the SSL session together, right after
      // the start messages have been exchanged.
      session.stopEncryption();
      /*
       * The second phase: the server reads this one before it sends its own, and registers
       * the handler once it has sent it. The list holds this peer and nothing else --
       * waitAndProcessTopoFromRemoteRS() reads rsInfos.get(0) above protocol version 4, so
       * an empty one ends the handshake on an IndexOutOfBoundsException instead, which is
       * an abort like any other and would leave the peer unregistered.
       */
      final RSInfo peerInfo = new RSInfo(PEER_RS_ID, registeredAs.toString(), -1, (byte) 1, 1);
      session.publish(new TopologyMsg(Collections.<DSInfo> emptyList(), newArrayList(peerInfo)));
      session.receive();
      return session;
    }
    catch (Exception e)
    {
      close(session);
      close(socket);
      throw e;
    }
  }

  /**
   * Answers, from the side of a peer replication server, the whole of the handshake this
   * server offers it, so that {@code ReplicationServerHandler.connect()} completes.
   * <p>
   * The mirror image of {@link #registerPeerFrom}: the initiator of a handshake sends the
   * first start message and the first topology message, so a peer which answers one reads
   * what that one publishes and publishes what it reads. The topology message holds this
   * peer and nothing else, for the reason it holds one there.
   *
   * @param session
   *          The session the handshake is running on.
   * @param received
   *          The {@code ReplServerStartMsg} the server under test sent.
   * @param answersOn
   *          The address this peer answers on, which is what its start message names and
   *          what it is therefore registered under.
   * @param baseDN
   *          The base DN of the domain the handshake is for.
   */
  private void answerWholeHandshake(Session session, ReplicationMsg received, HostPort answersOn, DN baseDN)
      throws Exception
  {
    session.publish(peerStartMsg(answersOn, baseDN));
    stopEncryptionWith(session, received);
    // The second phase: the server under test sends its topology message before it waits
    // for one, so this is read before the answer to it is published.
    session.receive();
    final RSInfo peerInfo = new RSInfo(PEER_RS_ID, answersOn.toString(), -1, (byte) 1, 1);
    session.publish(new TopologyMsg(Collections.<DSInfo> emptyList(), newArrayList(peerInfo)));
    /*
     * Read once more before the session is closed under this, as every answer here does:
     * the server under test registers the handler, reads the topology message above and
     * starts the session's threads after this peer has published it, and closing this end
     * inside that would abort the handshake this case needs completed. Returning once the
     * session has something on it, rather than holding it for good, is what keeps this peer
     * answering: it serves one connection at a time, and the connect thread dials it too.
     */
    session.receive();
  }

  /**
   * Waits for the domain of the provided server to hold a handler for the fake peer under
   * the provided address URL.
   * <p>
   * That handler is what makes the outbound handshake of the case reach
   * {@code ERR_DUPLICATE_REPLICATION_SERVER_ID}: the server ids match and the address URLs
   * do not. Without it the handshake ends on the second phase instead, which is an abort as
   * well and reports the same message -- the case would pass while pinning the wrong path.
   *
   * @param rs
   *          The server under test.
   * @param baseDN
   *          The base DN of the domain the peer registered with.
   * @param registeredAs
   *          The address URL the peer is expected to be registered under.
   * @param timeoutMs
   *          How long to wait for the registration, in milliseconds.
   */
  private void waitForRegistrationUnder(ReplicationServer rs, DN baseDN, HostPort registeredAs, long timeoutMs)
      throws Exception
  {
    final long deadline = System.currentTimeMillis() + timeoutMs;
    ReplicationServerHandler registered;
    while (true)
    {
      registered = rs.getReplicationServerDomain(baseDN).getConnectedRSs().get(PEER_RS_ID);
      if (registered != null || System.currentTimeMillis() > deadline)
      {
        break;
      }
      Thread.sleep(50);
    }
    assertThat(registered).as("the peer should have registered with the domain").isNotNull();
    assertThat(registered.getServerAddressURL())
        .as("the peer should be registered under the address its start message named, which is"
            + " not the one it is configured under")
        .isEqualTo(registeredAs.toString());
  }

  /**
   * Returns the start message of a fake peer, naming the provided address.
   * <p>
   * A generation id of -1 leaves the one of the domain alone: a positive one would be
   * adopted by the handshake, which is a change none of these tests is about.
   */
  private ReplServerStartMsg peerStartMsg(HostPort address, DN baseDN)
  {
    return new ReplServerStartMsg(PEER_RS_ID, address.toString(), baseDN, 100,
        new ServerState(), -1, false, (byte) 1, 5000);
  }

  /**
   * Leaves the SSL session when the sender of the provided start message does.
   * <p>
   * Both ends leave it together, and what a start message asks for is what its sender does
   * itself: reading the next message on the other stream would be reading a stream nothing
   * is written to.
   */
  private void stopEncryptionWith(Session session, ReplicationMsg startMsg) throws Exception
  {
    if (!((ReplServerStartMsg) startMsg).getSSLEncryption())
    {
      session.stopEncryption();
    }
  }

  /**
   * Reports a peer which is down, has every handshake it is offered answered by the
   * provided answer, and asserts that the outage was reported once, closed once, and
   * reported again once the peer was gone -- the last of which is only reachable because
   * the first was closed.
   * <p>
   * {@code connect()} is driven from the test thread so that each pass is one the test
   * names, but the connect thread of the server drives it too, for the same peer. What
   * makes the counts exact is not that the other thread is kept out of them: it is that
   * {@link ReplicationServer.ConnectFailureReporter} is idempotent in both directions, so
   * while the record is held every failure the connect thread adds is one the record
   * already holds, and while it is cleared every recovery it reports is one already
   * reported. The pass waited for below is what the counts do need, so that the
   * {@code retainAll} at the end of a pass which started without the domain cannot erase
   * the record they are about. The blacklist {@code runConnect()} keeps is local to it and
   * written only from its own failures, so the calls made here do not feed it.
   *
   * @param dbName
   *          The changelog directory of the server under test, which is its own.
   * @param answer
   *          How the peer answers the handshake this server offers it.
   */
  private void aPeerWhichAnswersClosesItsOutage(String dbName, PeerHandshake answer) throws Exception
  {
    TestCaseUtils.startServer();

    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final int[] ports = TestCaseUtils.findFreePorts(2);
    final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
    final String reachable =
        WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
    final String connected = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();

    final ReplicationServer[] servers = new ReplicationServer[1];
    final ExecutorService peerThread = Executors.newSingleThreadExecutor();
    try
    {
      final ReplSessionSecurity security = getReplSessionSecurity();
      final List<String> records = errorLogRecordsOf(live -> {
        servers[0] = new ReplicationServer(new ReplServerFakeConfiguration(
            ports[0], dbName, 0, RS_ID, 0, 100, newTreeSet(peerAddress.toString())));
        final ReplicationServer rs = servers[0];
        rs.getReplicationServerDomain(baseDN, true);
        /*
         * Wait for one whole pass of the connect thread before recording anything from
         * here. runConnect() snapshots the domains at the top of a pass and runs retainAll
         * against that snapshot at the bottom, so a pass which started before the domain
         * existed ends by clearing every record held for it -- including one this thread
         * would have made in between, which is the record the assertions below are about.
         */
        rs.waitConnections();

        assertThat(rs.connect(peerAddress, baseDN))
            .as("nothing listens on " + peerAddress).isFalse();
        waitForErrorLogRecord(live, connectErrorPrefix(peerAddress, baseDN), REPORT_TIMEOUT_MS);

        // The peer answers, and the handshake ends without a connection: the outage is over
        // and there is no session, which are two things rather than one.
        final AtomicBoolean serving = new AtomicBoolean(true);
        try (ServerSocket peerSocket = bindPeerPort(ports[1]))
        {
          peerSocket.setSoTimeout(SOCKET_TIMEOUT_MS);
          // Every connection is answered, not just the one below: the connect thread dials
          // the same peer, and a handshake left unanswered would time out rather than abort.
          peerThread.submit(() -> answerEveryHandshake(peerSocket, security, serving, answer));
          assertThat(rs.connect(peerAddress, baseDN))
              .as("a handshake which ends without a connection is not a connection, and a peer"
                  + " which answers with an abort every second is left alone for a few passes"
                  + " like one which does not answer at all").isFalse();
          waitForErrorLogRecord(live, reachable, REPORT_TIMEOUT_MS);
        }
        finally
        {
          serving.set(false);
        }

        // And is gone again, which is a new outage: the previous one was closed.
        assertThat(rs.connect(peerAddress, baseDN))
            .as("nothing listens on " + peerAddress).isFalse();
      });

      assertThat(countRecordsOf(records, reachable))
          .as("a peer which answers on its replication port ends the outage reported for it,"
              + " and this peer registers with no domain: the already connected branch of"
              + " runConnect() can report nothing about it, so connect() is the only place"
              + " this can have come from")
          .isEqualTo(1);
      assertThat(countRecordsOf(records, connected))
          .as("this peer has no replication session with this server, and reporting it"
              + " connected would leave that as the last thing said about a domain which"
              + " replicates nothing over it")
          .isEqualTo(0);
      assertThat(countRecordsOf(records, connectErrorPrefix(peerAddress, baseDN)))
          .as("the outage before the peer answered and the one after it went away again are"
              + " two outages, and the second is only reported because the first was closed:"
              + " a record left behind for a peer which answered silences it for good")
          .isEqualTo(2);
    }
    finally
    {
      peerThread.shutdownNow();
      remove(servers[0]);
    }
  }

  /**
   * Returns what every report of a failure to connect to the provided peer for the
   * provided domain starts with, which is the message up to the cause: the cause differs
   * between a peer which refuses a connection and one which resets it.
   */
  private String connectErrorPrefix(HostPort peer, DN baseDN)
  {
    final String message = WARN_REPLICATION_SERVER_CONNECT_ERROR.get(RS_ID, peer, baseDN, "").toString();
    return message.substring(0, message.indexOf(baseDN.toString()) + baseDN.toString().length());
  }

  /**
   * Returns where the first record holding the provided text sits in the provided error
   * log, for the assertions which are about the order of two records rather than about how
   * many there are: a count cannot tell a recovery which closed what a case is about from
   * one which closed something that happened in between.
   */
  private int indexOfRecord(List<String> records, String contained)
  {
    for (int i = 0; i < records.size(); i++)
    {
      if (records.get(i).contains(contained))
      {
        return i;
      }
    }
    throw new AssertionError(
        "the error log should hold \"" + contained + "\", but it holds: " + records);
  }

  /** Binds the provided port the way {@code TestCaseUtils} binds the ones it hands out. */
  private ServerSocket bindPeerPort(int port) throws Exception
  {
    final ServerSocket socket = new ServerSocket();
    socket.setReuseAddress(true);
    socket.bind(new InetSocketAddress(port));
    return socket;
  }

  /**
   * Answers the {@code ReplServerStartMsg} of every handshake offered to the provided
   * socket with the provided answer, until that socket is closed.
   * <p>
   * Every connection is answered rather than only the one a test drives: the connect thread
   * of the server dials the same peer, and a handshake left unanswered would end on the
   * connection timeout rather than on the answer, which is a different failure and a far
   * slower one.
   */
  private void answerEveryHandshake(ServerSocket peerSocket, ReplSessionSecurity security,
      AtomicBoolean serving, PeerHandshake answer)
  {
    while (serving.get() && !peerSocket.isClosed())
    {
      Socket accepted = null;
      Session session = null;
      try
      {
        accepted = peerSocket.accept();
        accepted.setTcpNoDelay(true);
        session = security.createServerSession(accepted, SOCKET_TIMEOUT_MS);
        answer.answer(session, session.receive());
      }
      catch (Exception ignored)
      {
        // Every answer ends by reading what the peer sends next, or does not send, and what
        // ends that read is the peer closing its own end. What ends this loop is the socket
        // being closed under the accept() above.
      }
      finally
      {
        close(session);
        close(accepted);
      }
    }
  }

  /** How a fake peer answers the {@code ReplServerStartMsg} a handshake starts with. */
  @FunctionalInterface
  private interface PeerHandshake
  {
    /**
     * Answers the provided start message on the provided session, and returns when the
     * handshake is over: the session is closed under it afterwards.
     *
     * @param session
     *          The session the handshake is running on.
     * @param received
     *          The {@code ReplServerStartMsg} the server under test sent.
     * @throws Exception
     *           When the session ends, which every answer here ends by waiting for.
     */
    void answer(Session session, ReplicationMsg received) throws Exception;
  }
}

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

@@ -22,6 +22,7 @@
import static org.testng.Assert.*;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
@@ -34,6 +35,8 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigChangeResult;
@@ -522,6 +525,423 @@
  }

  /**
   * Tests that the listen thread reports a failure of {@code accept()} and waits before
   * accepting again when it keeps failing on a socket which stays open, instead of
   * spinning on the failure in silence.
   * <p>
   * A process which ran out of file descriptors fails every {@code accept()} without ever
   * closing the listen socket, so the loop comes straight back to it. The wait is what
   * bounds that spin, and nothing but the time the loop takes tells a thread which waits
   * from one which does not.
   */
  @Test
  public void listenThreadReportsAcceptFailuresAndWaitsBeforeAcceptingAgain() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer replicationServer = null;
    try
    {
      final int[] ports = TestCaseUtils.findFreePorts(1);
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          ports[0], "listenThreadWaitsBeforeAcceptingAgainDb", 0, 1, 0, 0, null));
      final ReplicationServer listeningServer = replicationServer;

      // Four, so that two waits in a row are measured: a backoff timed from the previous
      // failure rather than from the previous wait grants one, and the failure which
      // follows it then looks isolated -- half of a continuous run goes unwaited, and only
      // a second interval in a row tells that apart.
      final int failures = 4;
      final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
      final AtomicInteger accepts = new AtomicInteger();
      // When each accept() was entered: the wait is between two of them, and measuring
      // the whole loop instead would put the cost of everything else in the same budget.
      final long[] acceptNanos = new long[failures];
      /*
       * A socket which fails every accept() and closes itself once it has failed enough
       * of them: the listen loop ends on a closed socket, which is what returns
       * runListen(). It closes itself for real rather than overriding isClosed(), which
       * ServerSocket.close() consults before closing anything up to Java 17: the port and
       * its file descriptor would be held for the rest of the test JVM.
       */
      final ServerSocket failingSocket = new ServerSocket(0)
      {
        @Override
        public Socket accept() throws IOException
        {
          final int attempt = accepts.incrementAndGet();
          acceptNanos[attempt - 1] = System.nanoTime();
          if (attempt >= failures)
          {
            super.close();
          }
          throw new IOException(acceptFailure);
        }
      };

      final List<String> records = errorLogRecordsOf(() -> {
        try
        {
          listeningServer.runListen(failingSocket);
        }
        finally
        {
          close(failingSocket);
        }
        return null;
      });

      assertTrue(failingSocket.isClosed(), "the socket the listen loop ended on should be closed");
      assertEquals(accepts.get(), failures, "the listen loop should have ended on the closed socket");
      /*
       * Of the four failures, the second and the third are waited on: the first of a row
       * is not, so that an isolated failure costs the connections behind it nothing, and
       * the fourth closes the socket, which ends the loop before it would wait.
       *
       * Compared in nanoseconds: converting to milliseconds first floors the measurement,
       * so a wait of exactly the backoff would read as one millisecond short of it.
       */
      final long backoffNanos = TimeUnit.MILLISECONDS.toNanos(ReplicationServer.ACCEPT_FAILURE_BACKOFF_MS);
      for (int attempt = 2; attempt <= 3; attempt++)
      {
        final long waitedNanos = acceptNanos[attempt] - acceptNanos[attempt - 1];
        assertTrue(waitedNanos >= backoffNanos,
            "the listen thread should have waited " + ReplicationServer.ACCEPT_FAILURE_BACKOFF_MS
                + " ms after failure " + attempt + " of a row before accepting again, but only "
                + TimeUnit.NANOSECONDS.toMicros(waitedNanos)
                + " microseconds passed before attempt " + (attempt + 1));
      }
      // And the first failure of the row is not waited on. Bounded by the backoff itself,
      // which is what waiting on it would cost: what separates the first two attempts is
      // one log record and the turn of the loop, three orders of magnitude below it.
      final long firstIntervalNanos = acceptNanos[1] - acceptNanos[0];
      assertTrue(firstIntervalNanos < backoffNanos,
          "the listen thread should not have waited after the first failure, so that an isolated"
              + " one costs the connections behind it nothing, but "
              + TimeUnit.NANOSECONDS.toMicros(firstIntervalNanos)
              + " microseconds passed between the first attempt and the second");

      int warnings = 0;
      int suppressed = 0;
      for (String record : records)
      {
        if (record.contains(acceptFailure))
        {
          // A suppressed failure is still recorded, with the information severity, which
          // the replication log publishes and the error log does not. This capture
          // publishes every severity, so it holds both.
          if (record.contains("severity=WARNING"))
          {
            warnings++;
            // The warning stands for the failures suppressed since the previous one, and
            // this is the first, so it stands for itself alone. Read out of the message
            // to pin which of its two numbers is the count and which is the interval.
            assertTrue(record.contains("every 5 minutes") && record.contains("(0 since the previous warning)"),
                "the warning should report the interval it is bounded by and the 0 failures it"
                    + " stands for, but it reads: " + record);
          }
          else
          {
            // Each suppressed record counts itself in what the next warning will stand
            // for, so they are numbered in the order they were written.
            suppressed++;
            assertTrue(record.contains("every 5 minutes")
                && record.contains("(" + suppressed + " since the previous warning)"),
                "suppressed record " + suppressed + " should report the interval it is bounded by"
                    + " and the " + suppressed + " failures counted so far, but it reads: " + record);
          }
        }
      }
      // The first failure is warned about, the second and the third are suppressed by the
      // throttle, and the fourth, on the closed socket, is how the loop is told to end.
      assertEquals(warnings, 1, "the listen thread should have warned about the first failure only,"
          + " but the error log holds " + warnings + " warnings about it: " + records);
      assertEquals(suppressed, 2, "the failures which follow should have been suppressed and kept,"
          + " but the error log holds " + suppressed + " suppressed records about them: " + records);
    }
    finally
    {
      remove(replicationServer);
    }
  }

  /**
   * Tests that a failure of {@code accept()} which follows a connection is not waited on,
   * and that a connection which cannot be turned into a session is reported.
   * <p>
   * The wait bounds a listen loop which is spinning on a failure in silence. A loop which
   * accepted a connection in between is doing work instead, and charging it a wait per
   * failure would make a stream of connections aborted between the handshake and
   * {@code accept()} -- a health check, a port scan -- pace the whole listen port. What
   * that gives up is the failure which alternates with a connection, under a process which
   * frees a file descriptor now and then: the accept it lets through resets the clock, and
   * the failure behind it is timed as isolated. Both halves are this line, so both are
   * pinned here rather than left to the comment above it.
   * <p>
   * The connection served is one no session can be built on, which is the other half of
   * what the listen loop reports: the accepted socket is closed, so setting its options
   * fails at once where a socket connected to nothing would spend the whole connection
   * timeout inside the SSL handshake. It was connected before it was closed, so it still
   * names the peer the report is about, which is what the report is read for here.
   */
  @Test
  public void listenThreadDoesNotWaitAfterAFailureWhichFollowedAConnection() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer replicationServer = null;
    ServerSocket connectedTo = null;
    try
    {
      final int[] ports = TestCaseUtils.findFreePorts(1);
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          ports[0], "listenThreadResetsBackoffOnAConnectionDb", 0, 2, 0, 0, null));
      final ReplicationServer listeningServer = replicationServer;

      connectedTo = new ServerSocket(0);
      final Socket served = new Socket();
      served.connect(new InetSocketAddress("127.0.0.1", connectedTo.getLocalPort()), 10000);
      final String servedAddress = served.getRemoteSocketAddress().toString();
      served.close();

      final int attempts = 4;
      final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
      final AtomicInteger accepts = new AtomicInteger();
      // When each accept() was entered: what is measured is between two of them.
      final long[] acceptNanos = new long[attempts];
      /*
       * Fails, serves one connection, fails again, and closes itself on the fourth attempt:
       * the listen loop ends on a closed socket, which is what returns runListen(). It
       * closes itself for real rather than overriding isClosed(), which
       * ServerSocket.close() consults before closing anything up to Java 17: the port and
       * its file descriptor would be held for the rest of the test JVM.
       */
      final ServerSocket failingSocket = new ServerSocket(0)
      {
        @Override
        public Socket accept() throws IOException
        {
          final int attempt = accepts.incrementAndGet();
          acceptNanos[attempt - 1] = System.nanoTime();
          if (attempt == 2)
          {
            return served;
          }
          if (attempt >= attempts)
          {
            super.close();
          }
          throw new IOException(acceptFailure);
        }
      };

      final List<String> records = errorLogRecordsOf(() -> {
        try
        {
          listeningServer.runListen(failingSocket);
        }
        finally
        {
          close(failingSocket);
        }
        return null;
      });

      assertTrue(failingSocket.isClosed(), "the socket the listen loop ended on should be closed");
      assertEquals(accepts.get(), attempts, "the listen loop should have ended on the closed socket");
      /*
       * The third attempt fails right after the second served a connection, so the fourth
       * follows it without a wait. Without the reset the third failure would be timed from
       * the first, which is a few microseconds behind it, and read as repeating it.
       *
       * Compared in nanoseconds: converting to milliseconds first floors the measurement,
       * so a wait of exactly the backoff would read as one millisecond short of it.
       */
      final long backoffNanos = TimeUnit.MILLISECONDS.toNanos(ReplicationServer.ACCEPT_FAILURE_BACKOFF_MS);
      final long afterConnectionNanos = acceptNanos[3] - acceptNanos[2];
      assertTrue(afterConnectionNanos < backoffNanos,
          "the listen thread should not have waited after a failure which followed a connection,"
              + " but " + TimeUnit.NANOSECONDS.toMicros(afterConnectionNanos)
              + " microseconds passed between the third attempt and the fourth");

      int sessionSetupWarnings = 0;
      for (String record : records)
      {
        if (record.contains("accepted a connection from " + servedAddress)
            && record.contains("severity=WARNING"))
        {
          sessionSetupWarnings++;
          // Read out of the message to pin which of its two numbers is the count and which
          // is the interval, and that the address is the peer rather than the listen port.
          assertTrue(record.contains("every 5 minutes") && record.contains("(0 since the previous warning)"),
              "the warning should report the interval it is bounded by and the 0 failures it"
                  + " stands for, but it reads: " + record);
        }
      }
      assertEquals(sessionSetupWarnings, 1, "the connection no session could be built on should"
          + " have been reported once, but the error log holds " + sessionSetupWarnings
          + " warnings about it: " + records);
    }
    finally
    {
      close(connectedTo);
      remove(replicationServer);
    }
  }

  /**
   * Tests that each pass of the listen loop bounds its own failures, both the ones of
   * {@code accept()} and the ones of the connections it accepts.
   * <p>
   * A listen port change runs a second listen thread -- {@code switchListenPort()} starts it
   * before it stops the one it replaces -- so the two overlap, and a five minute window
   * opened on the port which was left would suppress the first failure on the port which
   * replaced it: silence in {@code logs/errors} right after the administrator changed the
   * port to get out of trouble. What that asks of the code is that neither throttle be a
   * field of the server, and two passes of {@code runListen()} on one server is that
   * handover without the timing of it: with either throttle kept in a field, the second
   * pass reports nothing at all.
   */
  @Test
  public void eachListenPassBoundsItsOwnFailures() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer replicationServer = null;
    ServerSocket connectedTo = null;
    try
    {
      final int[] ports = TestCaseUtils.findFreePorts(1);
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          ports[0], "eachListenPassBoundsItsOwnFailuresDb", 0, 2, 0, 0, null));
      final ReplicationServer listeningServer = replicationServer;

      connectedTo = new ServerSocket(0);
      final int connectedToPort = connectedTo.getLocalPort();
      final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
      // The peers of the two connections, read before they are closed: a closed socket still
      // names the peer it was connected to, but only the one which was connected.
      final List<String> peers = new ArrayList<>();

      final List<String> records = errorLogRecordsOf(() -> {
        for (int pass = 0; pass < 2; pass++)
        {
          /*
           * Two connections per pass, not one: the first is reported and the second is the
           * one the throttle of the pass has to suppress. With a single connection a
           * throttle built for each failure rather than held for the pass reports the same
           * one warning, the first failure of a fresh throttle being a warning either way,
           * so the call site would be pinned by nothing.
           */
          final Socket reported = servedConnectionTo(connectedToPort);
          peers.add(reported.getRemoteSocketAddress().toString());
          final Socket suppressed = servedConnectionTo(connectedToPort);

          final AtomicInteger accepts = new AtomicInteger();
          /*
           * Serves those two connections, fails, and fails again on the socket it closes
           * under itself: two failures of one kind and one of the other per pass, and the
           * close is what returns runListen(). The failure which closes the socket is not
           * the reported one -- handleAcceptFailure() returns on a closed socket, that
           * failure being how a listen thread is told its port was taken away -- so the
           * pass has to fail once before it.
           */
          final ServerSocket failingSocket = new ServerSocket(0)
          {
            @Override
            public Socket accept() throws IOException
            {
              final int attempt = accepts.incrementAndGet();
              if (attempt == 1)
              {
                return reported;
              }
              if (attempt == 2)
              {
                return suppressed;
              }
              if (attempt >= 4)
              {
                super.close();
              }
              throw new IOException(acceptFailure);
            }
          };

          try
          {
            listeningServer.runListen(failingSocket);
          }
          finally
          {
            close(failingSocket);
          }
        }
        return null;
      });

      assertEquals(peers.size(), 2, "both passes should have served their connection");
      /*
       * Counted by the message rather than by the peer: the kernel hands the second
       * connection the ephemeral port the first one released, so the two passes usually
       * name the same peer, and one warning naming it twice is what the second pass
       * reporting its own connection looks like.
       */
      assertEquals(countWarningsOf(records, "accepted a connection from "), 2,
          "each pass should have reported the connection no session could be built on, but the"
              + " error log holds " + countWarningsOf(records, "accepted a connection from ")
              + " such warnings: " + records);
      for (String peer : peers)
      {
        assertTrue(countWarningsOf(records, "accepted a connection from " + peer) >= 1,
            "the report should name the peer of the connection rather than the listen port,"
                + " but no warning names " + peer + ": " + records);
      }
      assertEquals(countRecordsOf(records, "accepted a connection from "), 4,
          "each pass should have recorded both of its connections, the one it warned about and"
              + " the one its throttle suppressed, but the error log holds "
              + countRecordsOf(records, "accepted a connection from ") + " such records: " + records);
      assertEquals(countWarningsOf(records, acceptFailure), 2,
          "each pass should have warned about the failure of accept() which ended it, but the"
              + " error log holds " + countWarningsOf(records, acceptFailure) + " such warnings: "
              + records);
    }
    finally
    {
      close(connectedTo);
      remove(replicationServer);
    }
  }

  /**
   * Returns a socket connected to the provided port and closed, which is a connection no
   * replication session can be started on: setting the options of a closed socket fails at
   * once, where a socket connected to nothing would spend the whole connection timeout
   * inside the SSL handshake.
   */
  private Socket servedConnectionTo(int port) throws IOException
  {
    final Socket socket = new Socket();
    socket.connect(new InetSocketAddress("127.0.0.1", port), 10000);
    socket.close();
    return socket;
  }

  /** Returns how many of the provided error log records hold the provided text as warnings. */
  private int countWarningsOf(List<String> records, String contained)
  {
    int warnings = 0;
    for (String record : records)
    {
      if (record.contains(contained) && record.contains("severity=WARNING"))
      {
        warnings++;
      }
    }
    return warnings;
  }

  /**
   * Waits for the listen thread of the provided replication server and port to be inside
   * {@code accept()}, or to have left it.
   * <p>

 opendj-server-legacy/src/test/java/org/opends/server/util/FailureLogThrottleTest.java

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

import static java.util.concurrent.TimeUnit.MINUTES;
import static org.assertj.core.api.Assertions.assertThat;

import org.opends.server.DirectoryServerTestCase;
import org.testng.annotations.Test;

/** Tests for {@link FailureLogThrottle}. */
@SuppressWarnings("javadoc")
public class FailureLogThrottleTest extends DirectoryServerTestCase
{
  private static final long INTERVAL_NANOS = MINUTES.toNanos(5);

  private static FailureLogThrottle newThrottle()
  {
    return new FailureLogThrottle(5, MINUTES);
  }

  @Test
  public void theFirstFailureIsLoggedHoweverLongTheThrottleHasExisted() throws Exception
  {
    assertThat(newThrottle().record(System.nanoTime()))
        .as("the first failure is logged, and stands for itself alone").isEqualTo(0);
  }

  @Test
  public void failuresInsideTheIntervalAreSuppressedAndCounted() throws Exception
  {
    final FailureLogThrottle throttle = newThrottle();
    final long start = System.nanoTime();

    throttle.record(start);
    assertThat(throttle.record(start + 1))
        .as("the next failure is the first suppressed one").isEqualTo(-2);
    assertThat(throttle.record(start + INTERVAL_NANOS - 1))
        .as("a failure one nanosecond before the interval is up is the second").isEqualTo(-3);
  }

  @Test
  public void theNextFailureLoggedReportsTheSuppressedOnesBeforeIt() throws Exception
  {
    final FailureLogThrottle throttle = newThrottle();
    final long start = System.nanoTime();

    throttle.record(start);
    throttle.record(start + 1);
    throttle.record(start + 2);

    assertThat(throttle.record(start + INTERVAL_NANOS))
        .as("the failure ending the interval reports the two suppressed before it").isEqualTo(2);
    assertThat(throttle.record(start + 2 * INTERVAL_NANOS))
        .as("the count starts again from the failure last logged").isEqualTo(0);
  }

  @Test
  public void eachThrottleCountsOnItsOwn() throws Exception
  {
    final FailureLogThrottle throttle = newThrottle();
    final FailureLogThrottle otherThrottle = newThrottle();
    final long start = System.nanoTime();

    throttle.record(start);
    throttle.record(start + 1);

    assertThat(otherThrottle.record(start + 2))
        .as("a failure of another kind is logged with a count of its own").isEqualTo(0);
  }
}