[#1241] Log the benign aborts of a replication handshake as warnings: the c...

Valery Kharseko

22 hours ago 79e812a1ce0dc5e475a57a31c1a6ac2382c5b1a9

[#1241] Log the benign aborts of a replication handshake as warnings: the cross connection timeout (145) and a peer which went away (178, 179) (#1242)

Fixes #1241

## Problem

`ServerHandler.abortStart(LocalizableMessage)` logged the reason of
every aborted replication handshake through `logger.error`. Two kinds of
abort that are no error of this server came out at ERROR that way:

- msgID 145, `WARN_TIMEOUT_WHEN_CROSS_CONNECTION`: the listen thread
gives way to a cross connection on purpose. `lockDomainWithTimeout()`
threw it in a `DirectoryException`, and `startFromRemoteRS` passed it to
`abortStart` with the other `DirectoryException`s.
- msgID 178 and 179, `*_DISCONNECTED_DURING_HANDSHAKE`: an `IOException`
during the handshake, that is, the peer went away. A peer which gave up
on its own timeout logs that itself at WARNING (msgID 212), so one event
read as a WARNING on one side and an ERROR on the other.

A peer which restarts under load leaves both behind on every server it
was handshaking with (restart-ds rows of #1208, runs
[40463887505](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/40463887505)
and
[39909337627](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/39909337627)),
and an alert on `category=SYNC severity=ERROR` takes them for errors of
those servers.

## Fix

- `abortStart(reason)` still logs at ERROR. The new
`abortStartWithWarning(reason)` logs at WARNING; both run the same
private abort.
- `lockDomainWithTimeout()` returns whether it got the lock instead of
throwing. `startFromRemoteRS` aborts a timed out wait with
`abortStartWithWarning(WARN_TIMEOUT_WHEN_CROSS_CONNECTION)`, so the
`DirectoryException`s left in its catch are the rejections of this
server, still at ERROR.
- The three `IOException` catches of the handshake (`DataServerHandler`
and both directions of `ReplicationServerHandler`) abort with
`abortStartWithWarning`. In `DataServerHandler` the `IOException` is no
longer wrapped in a `DirectoryException` for the outer catch.
- A peer silent past the connection timeout fails `receive()` with a
`SocketTimeoutException`, an `IOException`, and is aborted on at WARNING
as well. In the first phase this is what the connect thread of a cross
connection reads while the listen thread of its peer waits up to 8 s for
the domain lock, against a connection timeout of 5 s by default. The
second phase used to log it as the raw "Read timed out" at ERROR.
- `waitAndProcessTopoFromRemoteRS()`, which reads the `TopologyMsg` of a
replication server in the second phase of the handshake, no longer wraps
an `IOException` of the session in a `DirectoryException`. A replication
server gone before it sent its topology was logged there as
`severity=ERROR msgID=-1 msg=no more data`; it is the 179 warning of the
first phase now.
- A frame too large to allocate is not the peer going away, though the
session reported it as an `IOException`. `Session.receive()` raises a
`PacketTooLargeException` for it, an `IOException` still for every other
caller, and the three catches abort on it ahead of their `IOException`
arm, at ERROR. The text is its own ("Packet too large, can't allocate N
bytes."), as the second phase of a replication server already logged it,
followed by the address of the peer (` Peer: host/ip:port`). Before this
PR the first phase and the handshake of a directory server logged that
frame under 178 or 179, which name the servers. `receive()` raises the
exception for any `OutOfMemoryError` while it reads or decodes a frame,
which is an error too.
- The keys of 178 and 179 are renamed from `ERR_` to `WARN_` in
`replication.properties` and in the five translations that have them
(de, es, fr, ja, zh_CN), as #906 did for 105. The message IDs and texts
stay as they are. The prefix is what the log message reference is built
from (`GenerateMessageFileMojo` lists the `ERR_` keys only), so the two
entries are removed from the committed `appendix-log-messages.adoc` as
well.

A protocol error (`ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE`,
"StartSessionMsg required"), a rejection by this server, a frame which
does not decode, a frame too large to allocate and an unexpected
exception are still logged at ERROR.

## Tests

`HandshakeAbortSeverityTest` drives each handler on a TLS session pair
of its own and reads the severity off the error log record of the
reason. The peer goes away through `Session.abort()`: `close()` would
send a `StopMsg`, which is a silent abort.

- `aCrossConnectionGivenWayToIsAWarning`: the listen thread waits for
the domain lock while another thread holds it. 145 is at WARNING, and no
179 follows it.
- `aDataServerGoneDuringTheHandshakeIsAWarning`,
`aReplicationServerGoneDuringTheHandshakeItStartedIsAWarning`,
`aReplicationServerGoneDuringTheHandshakeThisServerStartedIsAWarning`:
the peer aborts its end of the session before a directory server
handshake, before a handshake the peer started and before one this
server started; 178, 179 and 179 at WARNING.
-
`aReplicationServerGoneBeforeItsTopologyInTheHandshakeItStartedIsAWarning`,
`aReplicationServerGoneBeforeItsTopologyInTheHandshakeThisServerStartedIsAWarning`:
the peer reads what the handshake sent, then aborts while its topology
is due; 179 at WARNING.
- `aDataServerSilentDuringTheHandshakeIsAWarning`,
`aReplicationServerSilentDuringTheHandshakeThisServerStartedIsAWarning`,
`aReplicationServerSilentBeforeItsTopologyInTheHandshakeItStartedIsAWarning`:
the handler's session gets a 500 ms timeout and the peer stays silent,
in each of the three `IOException` catches; 178, 179 and 179 at WARNING.
- `aPeerWhichBreaksTheProtocolIsStillAnError`: the peer answers the
start message of this server with a `TopologyMsg`.
`aPeerWhichSendsNoTopologyInTheHandshakeItStartedIsAnError` and
`aPeerWhichSendsNoTopologyInTheHandshakeThisServerStartedIsAnError`: the
peer sends its start message again where its topology is due.
`aDataServerWhichSendsNoStartSessionIsAnError`: a directory server sends
its start message again where its start session is due ("Protocol error:
StartSessionMsg required."). All four protocol errors are still at
ERROR.
- `aServerIdAnotherReplicationServerHoldsIsAnError`: a peer names the
server id of a replication server the domain holds a session with at
another address; `ERR_DUPLICATE_REPLICATION_SERVER_ID` is still at
ERROR.
-
`aFrameTooLargeForTheStartMessageOfTheHandshakeThisServerStartedIsAnError`,
`aFrameTooLargeForTheTopologyInTheHandshakeItStartedIsAnError`,
`aFrameTooLargeForTheStartSessionOfADataServerIsAnError`: the peer sends
the length header `7fffffff` where a message is due. The text is at
ERROR, it names the peer, and there is no 178 or 179.
- `aFrameWhichDoesNotDecodeInTheHandshakeThisServerStartedIsAnError`:
the peer sends a frame of one byte, a message type no message has, where
the answer to the start message of this server is due. `receive()` fails
with a `DataFormatException`, and the `catch (Exception)` of `connect()`
logs "received message with unknown type" at ERROR.
- Every case of a directory server also asserts that the thread of the
session was never started. Only `finalizeStart()` starts it, after the
handler is registered with the domain, so this checks that the handshake
ends at the abort. It reads the thread, not the domain: the reader of a
handler started on a closed session may already have unregistered it.

## Verification

- **Red, on master** (round 0): the four original warning cases fail
with `severity=ERROR` in the same records as in the CI runs above; the
protocol error case passes.
- **Red, on master** (round 2): 12 of the 18 cases fail. They include
the three silent peer cases: 178 and 179 at ERROR, and in the second
phase `severity=ERROR msgID=-1 msg=Read timed out` instead of 179. The
other six pass: the four protocol errors, the rejection and the frame
which does not decode, which master logs at ERROR as well.
- **Red, on the round 0 head:** 4 of the 13 cases fail. Both
`…GoneBeforeItsTopology…` cases fail with `severity=ERROR msgID=-1
msg=no more data`, and the start message and start session frame cases
fail with 179/178 at WARNING.
- **Red, on the round 1 head** (c16e47b27e): 3 of the 18 cases fail, the
three frame cases, on the record which does not name the peer.
- **Mutants**, the whole class each time; each fails only its own cases:
- logging every abort as a warning (round 0):
`aPeerWhichBreaksTheProtocolIsStillAnError`;
  - no `return;` after 145: the cross connection case;
- either `DirectoryException` catch of `ReplicationServerHandler`
logging a warning: the `…SendsNoTopology…` case of its direction (and,
for `startFromRemoteRS`, the duplicate server id case);
- each of the three `PacketTooLargeException` catches removed: the frame
case of its handshake;
- the `IOException` wrapped again in `waitAndProcessTopoFromRemoteRS()`:
both `…GoneBeforeItsTopology…` cases;
- a `SocketTimeoutException` logged at ERROR, in each of the three
`IOException` catches: the silent peer case of that catch;
- no `return;` after the 178 warning of `startFromRemoteDS()`: the gone
and the silent directory server cases; no `return;` after its frame
error: the start session frame case. Both fail with `expected:<[NEW]>
but was:<[TERMINATED]>`: the handshake went on to `finalizeStart()`, and
the thread of the session had already ended by the time the case checked
it;
- "StartSessionMsg required" logged as a warning:
`aDataServerWhichSendsNoStartSessionIsAnError`;
- the `catch (Exception)` of `connect()` logging a warning:
`aFrameWhichDoesNotDecodeInTheHandshakeThisServerStartedIsAnError`.
- **Green:** reactor `-Pprecommit verify -pl opendj-server-legacy -am`
with `HandshakeAbortSeverityTest`, `HandshakeAbortRegistrationTest`,
`HandshakeAbortGenerationIdTest`, `ReplicationServerConnectFailureTest`
and `MultiHomedPeerTest`: 35 tests, no failure. The javadoc of
`opendj-server-legacy` builds with doclint on.

## Not covered

- The whole replication test package was not run locally; CI runs it.
- The `catch (Exception)` of
`ReplicationServerHandler.startFromRemoteRS()` is not pinned. A peer
reaches it only through a failure of the JDK: a topology without the
replication server which sent it fails on `getRsInfos().get(0)` with an
`IndexOutOfBoundsException`. That is a missing check from before this
PR, and a case would pin the text of the JDK.
- The `DirectoryException` catch of
`DataServerHandler.startFromRemoteDS()` is not pinned. The only reason
with a message that reaches it is `ERR_SESSION_STARTUP_INTERRUPTED`, a
thread interrupted in `finalizeStart()`. `ERR_RS_INVALID_INIT_STATUS` is
turned into a `DirectoryException` without a message by the inner `catch
(Exception)` and aborted silently, as it was before this PR.
- A replication server of an older version still logs 145, 178 and 179
at ERROR, so a topology that mixes versions keeps them until every
server runs this one.
- Once this is merged, #1233 no longer needs to let 145, 178 and 179
through in its restart window.

 opendj-doc-generated-ref/src/main/asciidoc/reference/appendix-log-messages.adoc

@@ -12,7 +12,7 @@
  information: "Portions copyright [year] [name of copyright owner]".
 
  Copyright 2017 ForgeRock AS.
  Portions Copyright 2024 3A Systems LLC.
  Portions Copyright 2024-2026 3A Systems LLC.
////

:figure-caption!:
@@ -12676,20 +12676,6 @@
+
Message: Could not retrieve the configuration for a replication domain matching the entry %s.

[#log-ref-log-ref-ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178]
ID: 178::
Severity: ERROR

+
Message: Directory server %s was attempting to connect to replication server %s but has disconnected in handshake phase. Error: %s.

[#log-ref-log-ref-ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179]
ID: 179::
Severity: ERROR

+
Message: Replication server %s was attempting to connect to replication server %s but has disconnected in handshake phase. Error: %s.

[#log-ref-log-ref-ERR_RS_BADLY_DISCONNECTED_181]
ID: 181::
Severity: ERROR

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

New file
@@ -0,0 +1,44 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.protocol;

import java.io.IOException;

/**
 * Raised by {@link Session#receive()} when this server runs out of heap for a frame: most
 * likely one whose length header announces more bytes than it can allocate, though an
 * {@link OutOfMemoryError} reading or decoding a frame of an ordinary size is raised as this too.
 * <p>
 * It is an {@link IOException}, so that every caller which takes it for a session to give up
 * still does. A handshake tells it apart from the other ones, which are the peer going away:
 * either the peer breaks the protocol, no handshake message coming near the size such a header
 * announces, or this server is short of heap, and either is an error.
 */
public class PacketTooLargeException extends IOException
{
  private static final long serialVersionUID = 4264358734106531071L;

  /**
   * Creates the exception for a frame of the provided length.
   *
   * @param length
   *          the length the header of the frame announced
   */
  public PacketTooLargeException(int length)
  {
    super("Packet too large, can't allocate " + length + " bytes.");
  }
}

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

@@ -738,7 +738,9 @@
   * need to implement any replication.
   *
   * @return The ReplicationMsg that was received.
   * @throws IOException When error happened during IO process.
   * @throws IOException When error happened during IO process, a
   *         {@link PacketTooLargeException} when this server runs out of heap
   *         for the frame received.
   * @throws DataFormatException When the data received is not formatted as a
   *         ReplicationMsg.
   * @throws NotSupportedOldVersionPDUException If the received PDU is part of
@@ -773,8 +775,7 @@
      }
      catch (final OutOfMemoryError e)
      {
        throw new IOException("Packet too large, can't allocate "
            + totalLength + " bytes.");
        throw new PacketTooLargeException(totalLength);
      }
    }
    catch (final IOException | DataFormatException | NotSupportedOldVersionPDUException | RuntimeException e)

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

@@ -40,6 +40,7 @@
import org.opends.server.replication.common.StatusMachine;
import org.opends.server.replication.common.StatusMachineEvent;
import org.opends.server.replication.protocol.ChangeStatusMsg;
import org.opends.server.replication.protocol.PacketTooLargeException;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartDSMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
@@ -465,11 +466,19 @@

        logStartSessionHandshake(inStartSessionMsg, outTopoMsg);
      }
      catch (PacketTooLargeException e)
      {
        // Not the peer going away, see PacketTooLargeException.
        logger.traceException(e);
        abortStart(LocalizableMessage.raw(
            e.getLocalizedMessage() + " Peer: " + session.getReadableRemoteAddress()));
        return;
      }
      catch(IOException e)
      {
        LocalizableMessage errMessage = ERR_DS_DISCONNECTED_DURING_HANDSHAKE.get(
            inServerStartMsg.getServerId(), replicationServer.getServerId());
        throw new DirectoryException(ResultCode.OTHER, errMessage);
        abortStartWithWarning(WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(
            inServerStartMsg.getServerId(), replicationServer.getServerId()));
        return;
      }
      catch (Exception e)
      {

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

@@ -770,19 +770,23 @@
     * 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
     * can be left to it. ReplicationServerHandler.connect() aborts at eight 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
     * about a domain which has no session for it. The remaining five 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().
     * this side of it -- the second logged as a warning, being no error of this server, and
     * the other two as errors. A peer silent past the connection timeout is of the second
     * kind. A frame this server has no heap for is of the first kind or of the third, though
     * the session reports it as an IOException, which the second kind is as well. 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. For a peer which names

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

@@ -37,6 +37,7 @@
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.protocol.PacketTooLargeException;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
@@ -160,7 +161,7 @@
   * 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
   * three of its eight 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
@@ -285,13 +286,20 @@
      super.finalizeStart();
      return true;
    }
    catch (PacketTooLargeException e)
    {
      // Not the peer going away, see PacketTooLargeException.
      logger.traceException(e);
      abortStart(LocalizableMessage.raw(
          e.getLocalizedMessage() + " Peer: " + session.getReadableRemoteAddress()));
      return false;
    }
    catch (IOException e)
    {
      logger.traceException(e);
      LocalizableMessage errMessage = ERR_RS_DISCONNECTED_DURING_HANDSHAKE.get(
      abortStartWithWarning(WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(
          getReplicationServerId(),
          session.getReadableRemoteAddress());
      abortStart(errMessage);
          session.getReadableRemoteAddress()));
      return false;
    }
    catch (DirectoryException e)
@@ -325,7 +333,13 @@
      // The initiator decides if the session is encrypted
      sslEncryption = processStartFromRemote(inReplServerStartMsg);

      lockDomainWithTimeout();
      if (!lockDomainWithTimeout())
      {
        // This server gives way to a cross connection, see lockDomainWithTimeout().
        abortStartWithWarning(WARN_TIMEOUT_WHEN_CROSS_CONNECTION.get(getBaseDN(), serverId,
            session.getReadableRemoteAddress(), getReplicationServerId()));
        return;
      }

      if (replicationServerDomain.isAlreadyConnectedToRS(this))
      {
@@ -414,10 +428,17 @@

      super.finalizeStart();
    }
    catch (PacketTooLargeException e)
    {
      // Not the peer going away, see PacketTooLargeException.
      logger.traceException(e);
      abortStart(LocalizableMessage.raw(
          e.getLocalizedMessage() + " Peer: " + session.getReadableRemoteAddress()));
    }
    catch (IOException e)
    {
      logger.traceException(e);
      abortStart(ERR_RS_DISCONNECTED_DURING_HANDSHAKE.get(
      abortStartWithWarning(WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(
          inReplServerStartMsg.getServerId(), replicationServer.getServerId()));
    }
    catch (DirectoryException e)
@@ -440,15 +461,22 @@
   * Wait receiving the TopologyMsg from the remote RS and process it.
   * @return the topologyMsg received or {@code null} if stop was received.
   * @throws DirectoryException
   * @throws IOException
   *           If the session failed, which the caller reports as it does a
   *           failure of the session in the first phase of the handshake.
   */
  private TopologyMsg waitAndProcessTopoFromRemoteRS()
      throws DirectoryException
      throws DirectoryException, IOException
  {
    ReplicationMsg msg;
    try
    {
      msg = session.receive();
    }
    catch (IOException e)
    {
      throw e;
    }
    catch(Exception e)
    {
      LocalizableMessage message = LocalizableMessage.raw(e.getLocalizedMessage());

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

@@ -205,11 +205,41 @@
  }

  /**
   * Abort a start procedure currently establishing.
   * Abort a start procedure currently establishing, logging the reason as an error.
   * @param reason The provided reason.
   */
  protected void abortStart(LocalizableMessage reason)
  {
    abortStart(reason, false);
  }

  /**
   * Abort a start procedure currently establishing, logging the reason as a warning.
   * <p>
   * For the aborts which are no error of this server and no fault of its peer
   * either: the peer going away while the handshake ran, which the peer logs
   * as a warning itself when it gave up on its own timeout, and this server
   * giving way to a cross connection. Both are what a peer restarting under
   * load leaves behind, and logged as errors they read as errors of every
   * server it was handshaking with.
   * <p>
   * The peer going away is an {@code IOException} of the session, in either
   * phase of the handshake. A peer silent past the connection timeout is one
   * too: it is also what the connect thread of a cross connection reads while
   * the listen thread of its peer waits for the domain lock. The exception is a
   * {@link org.opends.server.replication.protocol.PacketTooLargeException}:
   * a frame this server has no heap for is not the peer going away, and is an
   * error.
   *
   * @param reason The provided reason.
   */
  protected void abortStartWithWarning(LocalizableMessage reason)
  {
    abortStart(reason, true);
  }

  private void abortStart(LocalizableMessage reason, boolean isWarning)
  {
    // We did not recognize the message, close session as what can happen after
    // is undetermined and we do not want the server to be disturbed
    Session localSession = session;
@@ -221,7 +251,14 @@
        {
         logger.trace("In " + this + " closing session with err=" + reason);
        }
        logger.error(reason);
        if (isWarning)
        {
          logger.warn(reason);
        }
        else
        {
          logger.error(reason);
        }
      }

      // This method is only called when aborting a failing handshake and
@@ -790,25 +827,18 @@
   * thread to take the lock it was waiting for and process the connection
   * attempt.
   *
   * @throws DirectoryException
   *           When an exception occurs.
   * @return {@code true} if the lock was acquired, {@code false} if the wait
   *         timed out, which is the connection attempt to abort
   * @throws InterruptedException
   *           If the current thread was interrupted while waiting for the lock.
   */
  public void lockDomainWithTimeout() throws DirectoryException,
      InterruptedException
  public boolean lockDomainWithTimeout() throws InterruptedException
  {
    final Random random = new Random();
    final int randomTime = random.nextInt(6); // Random from 0 to 5
    // Wait at least 3 seconds + (0 to 5 seconds)
    final long timeout = 3000 + randomTime * 1000L;
    final boolean lockAcquired = replicationServerDomain.tryLock(timeout);
    if (!lockAcquired)
    {
      LocalizableMessage message = WARN_TIMEOUT_WHEN_CROSS_CONNECTION.get(
          getBaseDN(), serverId, session.getReadableRemoteAddress(), getReplicationServerId());
      throw new DirectoryException(ResultCode.OTHER, message);
    }
    return replicationServerDomain.tryLock(timeout);
  }

  /**

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

@@ -285,10 +285,10 @@
 for importing suffix %s data has been stopped due to fractional configuration \
 inconsistency : imported data set has some fractional configuration but not \
 local server
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=Directory server %s was \
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=Directory server %s was \
 attempting to connect to replication server %s but has disconnected in \
 handshake phase
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=Replication server %s was \
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=Replication server %s was \
 attempting to connect to replication server %s but has disconnected in \
 handshake phase
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=Error when loading a virtual \

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

@@ -12,6 +12,7 @@
#
# Copyright 2006-2010 Sun Microsystems, Inc.
# Portions Copyright 2011-2016 ForgeRock AS.
# Portions Copyright 2026 3A Systems, LLC.

# This file contains the primary Directory Server configuration.  It must not
# be directly edited while the server is online.  The server configuration
@@ -148,8 +149,8 @@
ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=Die Konfiguration f\u00fcr eine Replikationsdom\u00e4ne, die mit dem Eintrag %s \u00fcbereinstimmt, ist fehlgeschlagen
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_BAD_DATA_SET_176=Der LDIF-Import f\u00fcr die %s Suffix-Daten wurde aufgrund einer partiellen Konfigurationsinkonsistenz angehalten: Der importierte Datensatz besitzt nicht dieselbe Konfiguration wie der lokale Server
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_DATA_SET_IS_FRACTIONAL_177=Der LDIF-Import f\u00fcr die %s Suffix-Daten wurde aufgrund einer partiellen Konfigurationsinkonsistenz angehalten: Der importierte Datensatz besitzt dieselbe partielle Konfiguration, der lokale Server jedoch nicht
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=Directory-Server %s hat versucht, eine Verbindung zum Replikationsserver %s herzustellen, wurde aber in der Handshake-Phase getrennt
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=Replikationsserver %s hat versucht, eine Verbindung zum Replikationsserver %s herzustellen, wurde aber in der Handshake-Phase getrennt
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=Directory-Server %s hat versucht, eine Verbindung zum Replikationsserver %s herzustellen, wurde aber in der Handshake-Phase getrennt
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=Replikationsserver %s hat versucht, eine Verbindung zum Replikationsserver %s herzustellen, wurde aber in der Handshake-Phase getrennt
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=Fehler beim Laden eines virtuellen Attributs f\u00fcr externes \u00c4nderungsprotokoll: %s , Fehler: %s
NOTE_ERR_UNABLE_TO_ENABLE_ECL_183=Fehler in %s beim Aktivieren des externen \u00c4nderungsprotokolls: %s
ERR_RESYNC_REQUIRED_UNKNOWN_DOMAIN_IN_PROVIDED_COOKIE_185=Vollst\u00e4ndige Neusynchronisierung erforderlich. Ursache: Das angegebene Cookie enth\u00e4lt eine unbekannte replizierte Dom\u00e4ne %s. Aktuelle ab Cookie ist <%s>

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

@@ -12,6 +12,7 @@
#
# Copyright 2006-2010 Sun Microsystems, Inc.
# Portions Copyright 2011-2016 ForgeRock AS.
# Portions Copyright 2026 3A Systems, LLC.

# This file contains the primary Directory Server configuration.  It must not
# be directly edited while the server is online.  The server configuration
@@ -148,8 +149,8 @@
ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=No pudo recuperarse la configuraci\u00f3n para un dominio de repetici\u00f3n que coincida con la entrada %s
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_BAD_DATA_SET_176=La importaci\u00f3n de LDIF para importar los datos del sufijo %s se ha detenido debido a una inconsistencia de la configuraci\u00f3n de repetici\u00f3n fraccional: el conjunto de datos importados no tiene la misma configuraci\u00f3n fraccional que el servidor local
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_DATA_SET_IS_FRACTIONAL_177=La importaci\u00f3n de LDIF para importar los datos del sufijo %s se ha detenido debido a una inconsistencia de la configuraci\u00f3n de repetici\u00f3n fraccional: el conjunto de datos importados tiene alguna configuraci\u00f3n fraccional, pero no el servidor local
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=El Servidor de directorios %s ha tratado de conectarse al servidor de repetici\u00f3n %s, pero se ha desconectado en la fase de protocolo de enlace
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=El Servidor de repetici\u00f3n %s ha tratado de conectarse al servidor de repetici\u00f3n %s, pero se ha desconectado en la fase de protocolo de enlace
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=El Servidor de directorios %s ha tratado de conectarse al servidor de repetici\u00f3n %s, pero se ha desconectado en la fase de protocolo de enlace
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=El Servidor de repetici\u00f3n %s ha tratado de conectarse al servidor de repetici\u00f3n %s, pero se ha desconectado en la fase de protocolo de enlace
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=Error al cargar un atributo virtual para el registro de cambios externo: Atributo: %s, Error: %s
NOTE_ERR_UNABLE_TO_ENABLE_ECL_183=Error en %s al habilitar el registro de cambios externo: %s
ERR_RESYNC_REQUIRED_UNKNOWN_DOMAIN_IN_PROVIDED_COOKIE_185=Se requiere una resincronizaci\u00f3n completa. Raz\u00f3n: la cookie proporcionada contiene un dominio repetido desconocido %s. La cookie actual de partida es <%s>

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

@@ -12,6 +12,7 @@
#
# Copyright 2006-2010 Sun Microsystems, Inc.
# Portions Copyright 2011-2016 ForgeRock AS.
# Portions Copyright 2026 3A Systems, LLC.

# This file contains the primary Directory Server configuration.  It must not
# be directly edited while the server is online.  The server configuration
@@ -148,8 +149,8 @@
ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=Impossible de r\u00e9cup\u00e9rer la configuration pour un domaine de r\u00e9plication correspondant \u00e0 l'entr\u00e9e %s
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_BAD_DATA_SET_176=L\u2019importation LDIF des donn\u00e9es %s de suffixe a \u00e9t\u00e9 interrompue en raison d\u2019une incoh\u00e9rence de configuration partielle\u00a0: l\u2019ensemble de donn\u00e9es import\u00e9es n\u2019a pas la m\u00eame configuration partielle que le serveur local
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_DATA_SET_IS_FRACTIONAL_177=L\u2019importation LDIF des donn\u00e9es %s de suffixe a \u00e9t\u00e9 interrompue en raison d\u2019une incoh\u00e9rence de configuration partielle\u00a0: l\u2019ensemble de donn\u00e9es import\u00e9es a une configuration partielle mais le serveur local pas
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=Le serveur d'annuaire %s a tent\u00e9 de se connect\u00e9 au serveur de r\u00e9plication %s mais s'est d\u00e9connect\u00e9 lors de la phase de n\u00e9gociation (handshake)
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=Le serveur de r\u00e9plication %s a tent\u00e9 de se connect\u00e9 au serveur de r\u00e9plication %s mais s'est d\u00e9connect\u00e9 lors de la phase de n\u00e9gociation (handshake)
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=Le serveur d'annuaire %s a tent\u00e9 de se connect\u00e9 au serveur de r\u00e9plication %s mais s'est d\u00e9connect\u00e9 lors de la phase de n\u00e9gociation (handshake)
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=Le serveur de r\u00e9plication %s a tent\u00e9 de se connect\u00e9 au serveur de r\u00e9plication %s mais s'est d\u00e9connect\u00e9 lors de la phase de n\u00e9gociation (handshake)
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=Erreur lors du chargement d'un attribut virtuel pour le journal des modifications externe\u00a0: attribut\u00a0: %s, erreur\u00a0: %s
NOTE_ERR_UNABLE_TO_ENABLE_ECL_183=Erreur de %s lors de l'activation du journal des modifications externe\u00a0: %s
ERR_RESYNC_REQUIRED_UNKNOWN_DOMAIN_IN_PROVIDED_COOKIE_185=Resynchronisation compl\u00e8te requise. Raison\u00a0: le cookie fourni contient un domaine r\u00e9pliqu\u00e9 inconnu (%s). Le cookie actuel de base est <%s>

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

@@ -12,6 +12,7 @@
#
# Copyright 2006-2010 Sun Microsystems, Inc.
# Portions Copyright 2011-2016 ForgeRock AS.
# Portions Copyright 2026 3A Systems, LLC.

# This file contains the primary Directory Server configuration.  It must not
# be directly edited while the server is online.  The server configuration
@@ -147,8 +148,8 @@
ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=\u30a8\u30f3\u30c8\u30ea %s \u306b\u4e00\u81f4\u3059\u308b\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30c9\u30e1\u30a4\u30f3\u306e\u69cb\u6210\u3092\u53d6\u5f97\u3067\u304d\u307e\u305b\u3093\u3067\u3057\u305f
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_BAD_DATA_SET_176=\u30b5\u30d5\u30a3\u30c3\u30af\u30b9 %s \u30c7\u30fc\u30bf\u3092\u30a4\u30f3\u30dd\u30fc\u30c8\u3059\u308b\u305f\u3081\u306e LDIF \u30a4\u30f3\u30dd\u30fc\u30c8\u304c\u3001\u90e8\u5206\u69cb\u6210\u306e\u4e0d\u4e00\u81f4\u304c\u539f\u56e0\u3067\u4e2d\u6b62\u3055\u308c\u307e\u3057\u305f: \u30a4\u30f3\u30dd\u30fc\u30c8\u3055\u308c\u305f\u30c7\u30fc\u30bf\u30bb\u30c3\u30c8\u306b\u306f\u30ed\u30fc\u30ab\u30eb\u30b5\u30fc\u30d0\u30fc\u3068\u540c\u3058\u90e8\u5206\u69cb\u6210\u304c\u5b58\u5728\u3057\u307e\u305b\u3093
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_DATA_SET_IS_FRACTIONAL_177=\u30b5\u30d5\u30a3\u30c3\u30af\u30b9 %s \u30c7\u30fc\u30bf\u3092\u30a4\u30f3\u30dd\u30fc\u30c8\u3059\u308b\u305f\u3081\u306e LDIF \u30a4\u30f3\u30dd\u30fc\u30c8\u304c\u3001\u90e8\u5206\u69cb\u6210\u306e\u4e0d\u4e00\u81f4\u304c\u539f\u56e0\u3067\u4e2d\u6b62\u3055\u308c\u307e\u3057\u305f: \u30a4\u30f3\u30dd\u30fc\u30c8\u3055\u308c\u305f\u30c7\u30fc\u30bf\u30bb\u30c3\u30c8\u306b\u306f\u3044\u304f\u3064\u304b\u306e\u90e8\u5206\u69cb\u6210\u304c\u5b58\u5728\u3057\u307e\u3059\u304c\u3001\u30ed\u30fc\u30ab\u30eb\u30b5\u30fc\u30d0\u30fc\u306f\u3042\u308a\u307e\u305b\u3093
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=\u30c7\u30a3\u30ec\u30af\u30c8\u30ea\u30b5\u30fc\u30d0\u30fc %s \u306f\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306b\u63a5\u7d9a\u3057\u3088\u3046\u3068\u3057\u307e\u3057\u305f\u304c\u3001\u30cf\u30f3\u30c9\u30b7\u30a7\u30fc\u30af\u30d5\u30a7\u30fc\u30ba\u3067\u63a5\u7d9a\u89e3\u9664\u3055\u308c\u307e\u3057\u305f
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306f\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306b\u63a5\u7d9a\u3057\u3088\u3046\u3068\u3057\u307e\u3057\u305f\u304c\u3001\u30cf\u30f3\u30c9\u30b7\u30a7\u30fc\u30af\u30d5\u30a7\u30fc\u30ba\u3067\u63a5\u7d9a\u89e3\u9664\u3055\u308c\u307e\u3057\u305f
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=\u30c7\u30a3\u30ec\u30af\u30c8\u30ea\u30b5\u30fc\u30d0\u30fc %s \u306f\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306b\u63a5\u7d9a\u3057\u3088\u3046\u3068\u3057\u307e\u3057\u305f\u304c\u3001\u30cf\u30f3\u30c9\u30b7\u30a7\u30fc\u30af\u30d5\u30a7\u30fc\u30ba\u3067\u63a5\u7d9a\u89e3\u9664\u3055\u308c\u307e\u3057\u305f
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306f\u30ec\u30d7\u30ea\u30b1\u30fc\u30b7\u30e7\u30f3\u30b5\u30fc\u30d0\u30fc %s \u306b\u63a5\u7d9a\u3057\u3088\u3046\u3068\u3057\u307e\u3057\u305f\u304c\u3001\u30cf\u30f3\u30c9\u30b7\u30a7\u30fc\u30af\u30d5\u30a7\u30fc\u30ba\u3067\u63a5\u7d9a\u89e3\u9664\u3055\u308c\u307e\u3057\u305f
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=\u5916\u90e8\u66f4\u65b0\u5c65\u6b74\u30ed\u30b0\u306e\u4eee\u60f3\u5c5e\u6027\u3092\u8aad\u307f\u8fbc\u307f\u4e2d\u306b\u30a8\u30e9\u30fc\u304c\u767a\u751f\u3057\u307e\u3057\u305f: \u5c5e\u6027: %s\u3001\u30a8\u30e9\u30fc: %s
NOTE_ERR_UNABLE_TO_ENABLE_ECL_183=\u5916\u90e8\u66f4\u65b0\u5c65\u6b74\u30ed\u30b0\u3092\u6709\u52b9\u306b\u3059\u308b\u3068\u304d\u306b\u3001%s \u3067\u30a8\u30e9\u30fc\u304c\u767a\u751f\u3057\u307e\u3057\u305f: %s
ERR_RESYNC_REQUIRED_UNKNOWN_DOMAIN_IN_PROVIDED_COOKIE_185=\u5b8c\u5168\u306a\u518d\u540c\u671f\u304c\u5fc5\u8981\u3067\u3059\u3002\u7406\u7531: \u63d0\u4f9b\u3055\u308c\u305f Cookie \u306b\u542b\u307e\u308c\u308b\u3001\u30ec\u30d7\u30ea\u30b1\u30fc\u30c8\u3055\u308c\u305f\u30c9\u30e1\u30a4\u30f3 %s \u304c\u4e0d\u660e\u3067\u3059\u3002Cookie <%s>

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

@@ -12,6 +12,7 @@
#
# Copyright 2006-2010 Sun Microsystems, Inc.
# Portions Copyright 2011-2016 ForgeRock AS.
# Portions Copyright 2026 3A Systems, LLC.

# This file contains the primary Directory Server configuration.  It must not
# be directly edited while the server is online.  The server configuration
@@ -146,8 +147,8 @@
ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=\u65e0\u6cd5\u68c0\u7d22\u4e0e\u6761\u76ee %s \u5339\u914d\u7684\u590d\u5236\u57df\u914d\u7f6e
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_BAD_DATA_SET_176=\u7531\u4e8e\u90e8\u5206\u914d\u7f6e\u4e0d\u4e00\u81f4\uff0c\u56e0\u6b64\u5bfc\u5165\u540e\u7f00 %s \u6570\u636e\u7684 LDIF \u5bfc\u5165\u5df2\u505c\u6b62\uff1a\u5bfc\u5165\u7684\u6570\u636e\u96c6\u4e0e\u672c\u5730\u670d\u52a1\u5668\u7684\u90e8\u5206\u914d\u7f6e\u4e0d\u540c
NOTE_ERR_LDIF_IMPORT_FRACTIONAL_DATA_SET_IS_FRACTIONAL_177=\u7531\u4e8e\u90e8\u5206\u914d\u7f6e\u4e0d\u4e00\u81f4\uff0c\u56e0\u6b64\u5bfc\u5165\u540e\u7f00 %s \u6570\u636e\u7684 LDIF \u5bfc\u5165\u5df2\u505c\u6b62\uff1a\u5bfc\u5165\u7684\u6570\u636e\u96c6\u5305\u542b\u672c\u5730\u670d\u52a1\u5668\u6240\u6ca1\u6709\u7684\u4e00\u4e9b\u90e8\u5206\u914d\u7f6e
ERR_DS_DISCONNECTED_DURING_HANDSHAKE_178=\u76ee\u5f55\u670d\u52a1\u5668 %s \u5c1d\u8bd5\u8fde\u63a5\u5230\u590d\u5236\u670d\u52a1\u5668 %s\uff0c\u4f46\u662f\u5728\u63e1\u624b\u9636\u6bb5\u4e2d\u65ad\u5f00\u8fde\u63a5
ERR_RS_DISCONNECTED_DURING_HANDSHAKE_179=\u590d\u5236\u670d\u52a1\u5668 %s \u5c1d\u8bd5\u8fde\u63a5\u5230\u590d\u5236\u670d\u52a1\u5668 %s\uff0c\u4f46\u662f\u5728\u63e1\u624b\u9636\u6bb5\u4e2d\u65ad\u5f00\u8fde\u63a5
WARN_DS_DISCONNECTED_DURING_HANDSHAKE_178=\u76ee\u5f55\u670d\u52a1\u5668 %s \u5c1d\u8bd5\u8fde\u63a5\u5230\u590d\u5236\u670d\u52a1\u5668 %s\uff0c\u4f46\u662f\u5728\u63e1\u624b\u9636\u6bb5\u4e2d\u65ad\u5f00\u8fde\u63a5
WARN_RS_DISCONNECTED_DURING_HANDSHAKE_179=\u590d\u5236\u670d\u52a1\u5668 %s \u5c1d\u8bd5\u8fde\u63a5\u5230\u590d\u5236\u670d\u52a1\u5668 %s\uff0c\u4f46\u662f\u5728\u63e1\u624b\u9636\u6bb5\u4e2d\u65ad\u5f00\u8fde\u63a5
NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR_182=\u4e3a\u5916\u90e8\u66f4\u6539\u65e5\u5fd7\u52a0\u8f7d\u865a\u62df\u5c5e\u6027\u65f6\u51fa\u73b0\u9519\u8bef\uff1a\u5c5e\u6027: %s\uff0c\u9519\u8bef: %s
NOTE_ERR_UNABLE_TO_ENABLE_ECL_183=\u542f\u7528\u5916\u90e8\u66f4\u6539\u65e5\u5fd7\u65f6\u5728 %s \u4e2d\u51fa\u73b0\u9519\u8bef: %s
ERR_RESYNC_REQUIRED_UNKNOWN_DOMAIN_IN_PROVIDED_COOKIE_185=\u8981\u6c42\u5b8c\u5168\u518d\u540c\u6b65\u3002\u539f\u56e0\uff1a\u63d0\u4f9b\u7684 cookie \u5305\u542b\u672a\u77e5\u7684\u590d\u5236\u57df %s\u3002Cookie <%s>

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

New file
@@ -0,0 +1,846 @@
/*
 * 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.testng.Assert.*;

import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collections;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLSocket;

import org.forgerock.i18n.LocalizableMessage;
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.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.ServerStartMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.TopologyMsg;
import org.testng.annotations.Test;

/**
 * Tests for issue #1241: the severity an aborted replication handshake is logged with.
 * <p>
 * {@code abortStart} used to log the reason of every abort at ERROR, so a peer which
 * restarts under load reads as an error of every server it was handshaking with: the
 * listen thread which gives way to a cross connection
 * ({@code WARN_TIMEOUT_WHEN_CROSS_CONNECTION}, a warning by its own name), and the
 * handshake its peer left ({@code *_DISCONNECTED_DURING_HANDSHAKE}, where a peer which gave
 * up on its own timeout logs that timeout at WARNING itself). Both are warnings now, in
 * either phase of the handshake, and so is a peer which stays silent past the connection
 * timeout: this server gives up on it as it does on a peer which went away.
 * <p>
 * What this server rejects, a peer which breaks the protocol, a frame which does not decode
 * and a frame too large to allocate are still errors. Each call site picks the severity of its
 * own abort, so a case pins only the one it reaches: the cases which say so reach every
 * {@code abortStart} of a reason in the handshakes of a replication server and of a data
 * server but two, so that logging an abort as a warning is not a way to pass them. The two are
 * the {@code catch (Exception)} of {@code startFromRemoteRS()}, which a peer reaches only
 * through a failure of the JDK (a topology without the replication server which sent it fails
 * on {@code get(0)}), and the {@code catch (DirectoryException)} of
 * {@code startFromRemoteDS()}, which no peer reaches with a message.
 * <p>
 * Each test drives one handler on a session pair of its own, as the listen thread or the
 * connect thread of the server would, and reads the severity off the record of the reason.
 * The peer which goes away aborts its end of the session rather than closing it:
 * {@code Session.close()} publishes a {@code StopMsg}, which is a peer ending the handshake
 * in order, and is aborted on silently.
 */
@SuppressWarnings("javadoc")
public class HandshakeAbortSeverityTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  /** The connection timeout of a handshake whose peer stays silent: nothing comes before it. */
  private static final int SILENT_PEER_TIMEOUT_MS = 500;
  /** Longer than the longest wait of {@code lockDomainWithTimeout()}, which is 8 seconds. */
  private static final long LOCK_TIMEOUT_MS = 30000;
  private static final int REMOTE_DS_ID = 61;
  private static final int REMOTE_RS_ID = 62;
  private static final String REMOTE_RS_URL = "127.0.0.1:8989";
  /** What {@code Session.receive()} reports of the frame {@link #sendTooLargeFrameHeader} announces. */
  private static final LocalizableMessage PACKET_TOO_LARGE = LocalizableMessage.raw(
      "Packet too large, can't allocate " + Integer.MAX_VALUE + " bytes.");

  /**
   * The listen thread waits a random 3 to 8 seconds for the domain lock, so that of two
   * replication servers connecting to each other at the same time one gives way: the
   * connect thread holding the lock here is the one which goes on.
   */
  @Test
  public void aCrossConnectionGivenWayToIsAWarning() throws Exception
  {
    final int rsId = 8261;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityCrossDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      // The lock is held by a thread of its own, as the connect thread holds it: the
      // domain lock is reentrant, so the thread of the handshake would take it.
      final ExecutorService lockHolder = Executors.newSingleThreadExecutor();
      final List<String> records;
      try
      {
        final CountDownLatch locked = new CountDownLatch(1);
        final CountDownLatch handshakeDone = new CountDownLatch(1);
        final Future<Void> holding = lockHolder.submit(() -> {
          domain.lock();
          try
          {
            locked.countDown();
            handshakeDone.await();
          }
          finally
          {
            domain.release();
          }
          return null;
        });
        assertTrue(locked.await(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS),
            "the domain lock should have been taken");

        try
        {
          records = errorLogRecordsOf(() -> {
            handler.startFromRemoteRS(startMsg);
            return null;
          });
        }
        finally
        {
          handshakeDone.countDown();
        }
        holding.get(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      }
      finally
      {
        lockHolder.shutdownNow();
      }

      assertThat(recordOf(records, WARN_TIMEOUT_WHEN_CROSS_CONNECTION.get(
              baseDN, REMOTE_RS_ID, session.getReadableRemoteAddress(), rsId)))
          .as("giving way to a cross connection is a warning, as its message says")
          .contains("severity=WARNING");
      assertThat(countRecordsOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId).toString()))
          .as("the handshake ends where this server gives way to the cross connection")
          .isZero();
    });
  }

  @Test
  public void aDataServerGoneDuringTheHandshakeIsAWarning() throws Exception
  {
    final int rsId = 8262;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      remoteEnd.abort(new IOException("the directory server gave up on the handshake"));
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId)))
          .as("a directory server which went away during the handshake is a warning")
          .contains("severity=WARNING");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /**
   * A peer which stays connected and says nothing past the connection timeout is given up on
   * as one which went away.
   */
  @Test
  public void aDataServerSilentDuringTheHandshakeIsAWarning() throws Exception
  {
    final int rsId = 8274;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentDSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId)))
          .as("a directory server silent past the connection timeout is a warning")
          .contains("severity=WARNING");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /**
   * A directory server which sends something else where its start session message is due
   * breaks the protocol, and that is still an error.
   */
  @Test
  public void aDataServerWhichSendsNoStartSessionIsAnError() throws Exception
  {
    final int rsId = 8275;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityNoStartSessionDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(serverStartMsg(baseDN));
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, LocalizableMessage.raw("Protocol error: StartSessionMsg required.")))
          .as("a directory server which breaks the protocol is an error")
          .contains("severity=ERROR");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  @Test
  public void aReplicationServerGoneDuringTheHandshakeItStartedIsAWarning() throws Exception
  {
    final int rsId = 8263;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      remoteEnd.abort(new IOException("the replication server gave up on the handshake"));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server which went away during the handshake it started is a warning")
          .contains("severity=WARNING");
    });
  }

  @Test
  public void aReplicationServerGoneDuringTheHandshakeThisServerStartedIsAWarning() throws Exception
  {
    final int rsId = 8264;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      remoteEnd.abort(new IOException("the replication server went away"));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer went away, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server which went away during the handshake this server started"
              + " is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * What the connect thread of a cross connection reads, when the listen thread of its peer
   * waits for the domain lock longer than the connection timeout: up to 8 seconds against 5 by
   * default.
   */
  @Test
  public void aReplicationServerSilentDuringTheHandshakeThisServerStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8276;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentToRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer did not answer, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server silent past the connection timeout is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * The second phase of the handshake reads the topology of the peer through a method of its
   * own, which used to report every failure of the session as the error of a raw message.
   */
  @Test
  public void aReplicationServerGoneBeforeItsTopologyInTheHandshakeItStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8266;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        // Gone once it has read the start message of this server: that is sent, and the
        // handshake waits for the topology of the peer.
        goneAfterReading(1, remoteEnd, () -> {
          handler.startFromRemoteRS(startMsg);
          return null;
        });
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server which went away before it sent its topology is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * The case above for a peer which stays silent: the second phase used to log the timeout
   * as the error of the raw "Read timed out".
   */
  @Test
  public void aReplicationServerSilentBeforeItsTopologyInTheHandshakeItStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8277;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentFromRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server silent past the connection timeout before it sent its topology"
              + " is a warning")
          .contains("severity=WARNING");
    });
  }

  @Test
  public void aReplicationServerGoneBeforeItsTopologyInTheHandshakeThisServerStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8267;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        // Gone once it has read the start message and the topology of this server: both
        // are sent, and the handshake waits for the topology of the peer.
        assertFalse(goneAfterReading(2, remoteEnd, () -> handler.connect(baseDN, true)),
            "the peer went away, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server which went away before it sent its topology is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * A peer which answers the start message of this server with something else breaks the
   * protocol, and that is still an error: of the aborts, only the peer going away and the
   * cross connection are warnings.
   */
  @Test
  public void aPeerWhichBreaksTheProtocolIsStillAnError() throws Exception
  {
    final int rsId = 8265;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityProtocolDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(new TopologyMsg(
          Collections.<DSInfo> emptyList(), Collections.<RSInfo> emptyList()));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer broke the protocol, so this is no session");
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              TopologyMsg.class.getCanonicalName(), "ReplServerStartMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  /**
   * The case above in the second phase, which is aborted from the {@code DirectoryException}
   * arm of the handshake: a peer which sends its start message again where its topology is
   * due.
   */
  @Test
  public void aPeerWhichSendsNoTopologyInTheHandshakeItStartedIsAnError() throws Exception
  {
    final int rsId = 8268;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSNoTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              ReplServerStartMsg.class.getCanonicalName(), "TopologyMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  @Test
  public void aPeerWhichSendsNoTopologyInTheHandshakeThisServerStartedIsAnError() throws Exception
  {
    final int rsId = 8269;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSNoTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the messages they answer: the session reads them all the same.
      remoteEnd.publish(replServerStartMsg(baseDN));
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer broke the protocol, so this is no session");
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              ReplServerStartMsg.class.getCanonicalName(), "TopologyMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  /**
   * A rejection by this server: a peer which names the server id of a replication server the
   * domain already holds a session with, at another address.
   */
  @Test
  public void aServerIdAnotherReplicationServerHoldsIsAnError() throws Exception
  {
    final int rsId = 8270;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDuplicateDb", rsId, (replicationServer, firstEnd, first) -> {
      // The first peer completes its handshake, so that the domain holds it: its topology is
      // published ahead of the start message of this server it answers.
      final ReplServerStartMsg firstStartMsg =
          (ReplServerStartMsg) sendThrough(firstEnd, first, replServerStartMsg(baseDN));
      firstEnd.publish(new TopologyMsg(Collections.<DSInfo> emptyList(),
          Collections.singletonList(new RSInfo(REMOTE_RS_ID, REMOTE_RS_URL, -1, (byte) 1, 1))));
      final ReplicationServerHandler registered =
          new ReplicationServerHandler(first, 100, replicationServer, 100);
      registered.startFromRemoteRS(firstStartMsg);
      // The precondition of the case rather than an assumption.
      assertThat(replicationServer.getReplicationServerDomain(baseDN).getConnectedRSs().get(REMOTE_RS_ID))
          .as("the first peer should hold its session")
          .isSameAs(registered);

      try (ServerSocket listen = TestCaseUtils.bindFreePort())
      {
        listen.setSoTimeout(SOCKET_TIMEOUT_MS);
        final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
        try (Session remoteEnd = sessionPair[0];
            Session session = sessionPair[1])
        {
          final ReplServerStartMsg startMsg = (ReplServerStartMsg) sendThrough(remoteEnd, session,
              replServerStartMsg(baseDN, "127.0.0.1:8990"));
          final ReplicationServerHandler handler =
              new ReplicationServerHandler(session, 100, replicationServer, 100);

          final List<String> records = errorLogRecordsOf(() -> {
            handler.startFromRemoteRS(startMsg);
            return null;
          });

          assertThat(recordOf(records, ERR_DUPLICATE_REPLICATION_SERVER_ID.get(
                  replicationServer.getMonitorInstanceName(), registered.getServerAddressURL(),
                  handler.getServerAddressURL(), REMOTE_RS_ID)))
              .as("a peer this server rejects is an error")
              .contains("severity=ERROR");
        }
      }
    });
  }

  /**
   * A frame too large to allocate is not the peer going away, though the session reports it
   * as an {@code IOException}: either the peer breaks the protocol, a handshake message being
   * nowhere near that size, or this server has no heap to spare it, and both are errors.
   */
  @Test
  public void aFrameTooLargeForTheStartMessageOfTheHandshakeThisServerStartedIsAnError()
      throws Exception
  {
    final int rsId = 8271;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      sendTooLargeFrameHeader(remoteEnd);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the frame cannot be read, so this is no session");
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress()));
    });
  }

  /**
   * A frame which does not decode fails in {@code Session.receive()} with neither an
   * {@code IOException} nor a {@code DirectoryException}, and is aborted on from the
   * {@code catch (Exception)} of the handshake, still as an error.
   */
  @Test
  public void aFrameWhichDoesNotDecodeInTheHandshakeThisServerStartedIsAnError()
      throws Exception
  {
    final int rsId = 8278;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSUndecodableDb", rsId, (replicationServer, remoteEnd, session) -> {
      // A frame of one byte, a message type which no message has.
      sendRaw(remoteEnd, new byte[] { '0', '0', '0', '0', '0', '0', '0', '1', 0 });
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the frame cannot be decoded, so this is no session");
        return null;
      });

      assertThat(recordOf(records, LocalizableMessage.raw("received message with unknown type")))
          .as("a frame which does not decode is an error")
          .contains("severity=ERROR");
      assertThat(countRecordsOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())
                  .toString()))
          .as("a frame which does not decode is not the peer going away")
          .isZero();
    });
  }

  @Test
  public void aFrameTooLargeForTheTopologyInTheHandshakeItStartedIsAnError() throws Exception
  {
    final int rsId = 8272;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      sendTooLargeFrameHeader(remoteEnd);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId));
    });
  }

  @Test
  public void aFrameTooLargeForTheStartSessionOfADataServerIsAnError() throws Exception
  {
    final int rsId = 8273;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      sendTooLargeFrameHeader(remoteEnd);
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId));
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /** One case: the replication server under test, and a session pair to it. */
  private interface HandshakeCase
  {
    void run(ReplicationServer replicationServer, Session remoteEnd, Session session)
        throws Exception;
  }

  /**
   * Runs the provided case on a replication server of its own and a session pair to it,
   * which are torn down once it ran.
   */
  private void runCase(String dbName, int rsId, HandshakeCase handshakeCase) throws Exception
  {
    ReplicationServer replicationServer = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(dbName, rsId);

      final Session[] sessionPair =
          connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        handshakeCase.run(replicationServer, remoteEnd, session);
      }
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * Runs the provided handshake on a thread of its own, as the listen thread or the connect
   * thread of the server would, while the remote end reads the provided number of messages
   * and goes away: the handshake then fails on what it waits for next, rather than on a
   * message it sends.
   */
  private static <T> T goneAfterReading(int messages, Session remoteEnd, Callable<T> handshake)
      throws Exception
  {
    final ExecutorService handshakeThread = Executors.newSingleThreadExecutor();
    try
    {
      final Future<T> done = handshakeThread.submit(handshake);
      for (int i = 0; i < messages; i++)
      {
        remoteEnd.receive();
      }
      remoteEnd.abort(new IOException("the replication server went away"));
      return done.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    }
    finally
    {
      handshakeThread.shutdownNow();
    }
  }

  /**
   * Writes the length header of a frame of {@link Integer#MAX_VALUE} bytes from the remote
   * end, and nothing after it: more than any array can hold, so the local end fails to
   * allocate it whatever its heap. The session writes no frame it did not encode itself, so
   * the header goes to the socket under it.
   */
  private static void sendTooLargeFrameHeader(Session remoteEnd) throws Exception
  {
    sendRaw(remoteEnd, String.format("%08x", Integer.MAX_VALUE).getBytes());
  }

  /** Writes the provided bytes from the remote end to the socket under it, as they are. */
  private static void sendRaw(Session remoteEnd, byte[] bytes) throws Exception
  {
    final Field secureSocket = Session.class.getDeclaredField("secureSocket");
    secureSocket.setAccessible(true);
    final OutputStream output = ((SSLSocket) secureSocket.get(remoteEnd)).getOutputStream();
    output.write(bytes);
    output.flush();
  }

  private static void assertTooLargeFrameIsAnError(List<String> records, Session session,
      LocalizableMessage disconnected)
  {
    final String record = recordOf(records, PACKET_TOO_LARGE);
    assertThat(record)
        .as("a frame too large to allocate is an error")
        .contains("severity=ERROR");
    assertThat(record)
        .as("the record of a frame too large to allocate names the peer which sent it")
        .contains("Peer: " + session.getReadableRemoteAddress());
    assertThat(countRecordsOf(records, disconnected.toString()))
        .as("a frame too large to allocate is not the peer going away")
        .isZero();
  }

  /**
   * Asserts that the handshake of a data server ended where it was aborted: only
   * {@code finalizeStart()} starts the thread of the session, once the handler is registered
   * with the domain. Read off the thread rather than off the domain, which the reader of a
   * handler started on a closed session may have unregistered it from already.
   */
  private static void assertAbortedHandshakeWentNoFurther(Session session)
  {
    assertThat(session.getState())
        .as("a handshake which was aborted starts no session thread")
        .isEqualTo(Thread.State.NEW);
  }

  private ReplicationServer newReplicationServer(String dbName, int rsId) throws Exception
  {
    return new ReplicationServer(new ReplServerFakeConfiguration(
        TestCaseUtils.findFreePort(), dbName, 0, rsId, 0, 100, new TreeSet<String>()));
  }

  private ServerStartMsg serverStartMsg(DN baseDN)
  {
    return new ServerStartMsg(REMOTE_DS_ID, "127.0.0.1:1389", baseDN, 100, 0,
        new ServerState(), -1, true, (byte) 1);
  }

  private ReplServerStartMsg replServerStartMsg(DN baseDN)
  {
    return replServerStartMsg(baseDN, REMOTE_RS_URL);
  }

  /**
   * Returns the start message of a fake peer replication server.
   * <p>
   * A generation id of -1 leaves the one of the domain alone, and the session stays
   * encrypted, so that neither end has to leave the SSL session for the handshake to go on.
   */
  private ReplServerStartMsg replServerStartMsg(DN baseDN, String serverURL)
  {
    return new ReplServerStartMsg(REMOTE_RS_ID, serverURL, baseDN, 100,
        new ServerState(), -1, true, (byte) 1, 5000);
  }

  /**
   * Publishes the provided start message from the remote end and returns it as the local
   * end reads it, which is the message the listen thread hands to the handler: it carries
   * the protocol version of the session, which a message built here does not.
   */
  private ReplicationMsg sendThrough(Session remoteEnd, Session session, ReplicationMsg startMsg)
      throws Exception
  {
    remoteEnd.publish(startMsg);
    return session.receive();
  }

  /** Teardown must never mask the primary assertion failure. */
  private void removeQuietly(ReplicationServer replicationServer)
  {
    try
    {
      remove(replicationServer);
    }
    catch (Exception ignored)
    {
    }
  }

  /**
   * Establishes a connected session pair over the given listen socket, as a
   * remote server connecting to the RS would. The TLS negotiation performed by
   * the session factories needs both ends handshaking at the same time, so the
   * client end runs on its own thread.
   *
   * @return the two sessions: the remote (client) end first, then the local
   *         (server) end to hand to the handler under test
   */
  private Session[] connectSessionPair(ServerSocket listenSocket,
      final ReplSessionSecurity security) throws Exception
  {
    final Socket clientSocket =
        new Socket("127.0.0.1", listenSocket.getLocalPort());
    clientSocket.setTcpNoDelay(true);
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    try
    {
      final Future<Session> clientEnd = executor.submit(
          () -> security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS));

      final Socket serverSocket = listenSocket.accept();
      serverSocket.setTcpNoDelay(true);
      final Session serverEnd =
          security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS);
      assertNotNull(serverEnd,
          "could not create a session for the handler under test");

      return new Session[] {
        clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
    }
    finally
    {
      executor.shutdown();
    }
  }
}