From 79e812a1ce0dc5e475a57a31c1a6ac2382c5b1a9 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 28 Oct 2026 16:24:44 +0000
Subject: [PATCH] [#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)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java | 7
opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortSeverityTest.java | 846 +++++++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 4
opendj-server-legacy/src/messages/org/opends/messages/replication_zh_CN.properties | 5
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/PacketTooLargeException.java | 44 ++
opendj-server-legacy/src/messages/org/opends/messages/replication_es.properties | 5
opendj-doc-generated-ref/src/main/asciidoc/reference/appendix-log-messages.adoc | 16
opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java | 15
opendj-server-legacy/src/messages/org/opends/messages/replication_de.properties | 5
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java | 18
opendj-server-legacy/src/messages/org/opends/messages/replication_ja.properties | 5
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java | 42 +
opendj-server-legacy/src/messages/org/opends/messages/replication_fr.properties | 5
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java | 56 ++
14 files changed, 1,013 insertions(+), 60 deletions(-)
diff --git a/opendj-doc-generated-ref/src/main/asciidoc/reference/appendix-log-messages.adoc b/opendj-doc-generated-ref/src/main/asciidoc/reference/appendix-log-messages.adoc
index 95100bf..546eebb 100644
--- a/opendj-doc-generated-ref/src/main/asciidoc/reference/appendix-log-messages.adoc
+++ b/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
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/PacketTooLargeException.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/PacketTooLargeException.java
new file mode 100644
index 0000000..caa988a
--- /dev/null
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/PacketTooLargeException.java
@@ -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.");
+ }
+}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
index a7823e3..f272b34 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
+++ b/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)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java
index 4e4bac6..3f554a2 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java
+++ b/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)
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
index f83b4ed..4aa9a57 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
+++ b/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
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java
index 85f2b76..4af6f38 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java
+++ b/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());
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
index 6694a9f..a6961cf 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
+++ b/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);
}
/**
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index eb763ec..a5244ee 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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 \
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication_de.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication_de.properties
index 344a2cb..2efec0b 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication_de.properties
+++ b/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>
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication_es.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication_es.properties
index 93065db..8fdc8de 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication_es.properties
+++ b/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>
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication_fr.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication_fr.properties
index ff65901..8c41585 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication_fr.properties
+++ b/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>
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication_ja.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication_ja.properties
index 227d787..2d14d96 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication_ja.properties
+++ b/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>
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication_zh_CN.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication_zh_CN.properties
index 362ec24..adf0587 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication_zh_CN.properties
+++ b/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>
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortSeverityTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortSeverityTest.java
new file mode 100644
index 0000000..b9784ee
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortSeverityTest.java
@@ -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();
+ }
+ }
+}
--
Gitblit v1.10.0