From a77994116fdc36bff765e79f85fdb29027259f32 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 02 Nov 2026 08:21:08 +0000
Subject: [PATCH] [#1250] Report unexpected replication stream EOF as a bad disconnection (#1254)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java | 3
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java | 6
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ServerReaderDisconnectTest.java | 246 +++++++++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerReader.java | 18 +++
4 files changed, 269 insertions(+), 4 deletions(-)
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 f272b34..e95f929 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
@@ -22,6 +22,7 @@
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.Closeable;
+import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -795,7 +796,7 @@
if (read == -1)
{
lastReceiveTime = 0;
- throw new IOException("no more data");
+ throw new EOFException("no more data");
}
length += read;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerReader.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerReader.java
index 7eb1c9e..424c8c3 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerReader.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerReader.java
@@ -13,9 +13,11 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.server;
+import java.io.EOFException;
import java.net.SocketException;
import org.forgerock.i18n.LocalizableMessage;
@@ -206,6 +208,22 @@
logger.error(errMessage);
}
}
+ catch (EOFException e)
+ {
+ // V4 and later peers send StopMsg before an orderly close unless their session recorded an error.
+ if (!handler.shuttingDown() && !session.closeInitiated()
+ && session.getProtocolVersion() >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
+ {
+ logException(e);
+ errMessage = handler.getBadlyDisconnectedErrorMessage();
+ logger.error(errMessage);
+ }
+ else
+ {
+ errMessage = NOTE_READER_EXCEPTION.get(handler, stackTraceToSingleLineString(e));
+ logger.info(errMessage);
+ }
+ }
catch (Exception e)
{
/*
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java
index 89fe2f8..fce6760 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java
@@ -150,9 +150,9 @@
* handler closes it too - how the connection ended is what the reader saw, not this writer.
* The reader logs nothing for the StopMsg of a peer leaving in order, and reports a read
* which failed on the socket as a bad disconnection, at ERROR. An end of the stream with no
- * StopMsg before it - a peer older than protocol V4, a peer killed with nothing left unread,
- * a peer giving up a session which had failed - it reports only at NOTICE, as
- * NOTE_READER_EXCEPTION (issue #1250).
+ * StopMsg before it is reported as a bad disconnection for a peer using protocol V4 or
+ * later, unless the handler is shutting down or this side has begun the close; otherwise,
+ * and for an older peer, at NOTICE.
*
* And a session closed on this side is the only one this writer meets a failing write on
* while the session runs the publisher thread ServerHandler starts: publish() then queues, a
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ServerReaderDisconnectTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ServerReaderDisconnectTest.java
new file mode 100644
index 0000000..3e2721b
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ServerReaderDisconnectTest.java
@@ -0,0 +1,246 @@
+/*
+ * 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.
+ * Portions Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.server;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.opends.messages.ReplicationMessages.ERR_DS_BADLY_DISCONNECTED;
+import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+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.opendj.ldap.DN;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.protocol.ProtocolVersion;
+import org.opends.server.replication.protocol.Session;
+import org.opends.server.replication.protocol.StopMsg;
+import org.testng.annotations.Test;
+
+/** Checks the records emitted when a replication server reader loses its peer. */
+@SuppressWarnings("javadoc")
+public class ServerReaderDisconnectTest extends ReplicationTestCase
+{
+ @Test
+ public void eofFromModernPeerIsReported() throws Exception
+ {
+ assertThat(records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false))
+ .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
+ .hasSize(1);
+ }
+
+ @Test
+ public void eofFromCurrentPeerIsReported() throws Exception
+ {
+ assertThat(records(new byte[0], ProtocolVersion.getCurrentVersion(), false, false))
+ .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
+ .hasSize(1);
+ }
+
+ @Test
+ public void truncatedFrameFromModernPeerIsReported() throws Exception
+ {
+ assertThat(records("00000004x".getBytes(StandardCharsets.US_ASCII),
+ ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false))
+ .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
+ .hasSize(1);
+ }
+
+ @Test
+ public void eofFromOldPeerKeepsNotice() throws Exception
+ {
+ final List<String> logged = records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V3, false, false);
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
+ }
+
+ @Test
+ public void eofDuringShutdownIsNotReportedAsBadDisconnection() throws Exception
+ {
+ final List<String> logged = records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, true, false);
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
+ }
+
+ @Test
+ public void eofAfterLocalCloseIsNotReportedAsBadDisconnection() throws Exception
+ {
+ assertThat(records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, false, true))
+ .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
+ }
+
+ @Test
+ public void eofAfterConcurrentLocalCloseIsNotReportedAsBadDisconnection() throws Exception
+ {
+ final CountDownLatch reading = new CountDownLatch(1);
+ final CountDownLatch eof = new CountDownLatch(1);
+ final InputStream input = new InputStream()
+ {
+ @Override
+ public int read() throws IOException
+ {
+ reading.countDown();
+ try
+ {
+ assertThat(eof.await(10, TimeUnit.SECONDS)).as("local close released the read").isTrue();
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("Interrupted while waiting for EOF");
+ }
+ return -1;
+ }
+ };
+ try (ServerSocket listener = new ServerSocket(0);
+ Socket peer = new Socket("127.0.0.1", listener.getLocalPort());
+ Socket socket = listener.accept();
+ Session session = new Session(socket, new TestSocket(input)))
+ {
+ session.setProtocolVersion(ProtocolVersion.REPLICATION_PROTOCOL_V4);
+ final DataServerHandler handler = handler(session, false);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try
+ {
+ final List<String> logged = errorLogRecordsOf(() -> {
+ final Future<?> reader = executor.submit(() -> new ServerReader(session, handler).run());
+ try
+ {
+ assertThat(reading.await(10, TimeUnit.SECONDS)).as("reader entered receive").isTrue();
+ // Reading closeInitiated before receive() would miss this close and log a false ERROR.
+ session.close();
+ eof.countDown();
+ reader.get(10, TimeUnit.SECONDS);
+ }
+ finally
+ {
+ eof.countDown();
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).as("reader stopped").isTrue();
+ }
+ return null;
+ });
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ public void stopMessageIsNotReportedAsBadDisconnection() throws Exception
+ {
+ final byte[] message = new StopMsg().getBytes(ProtocolVersion.REPLICATION_PROTOCOL_V4);
+ final ByteArrayOutputStream frame = new ByteArrayOutputStream();
+ frame.write(String.format("%08x", message.length).getBytes(StandardCharsets.US_ASCII));
+ frame.write(message);
+ final List<String> logged = records(frame.toByteArray(), ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false);
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
+ assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).isEmpty();
+ }
+
+ private List<String> records(final byte[] input, final short version, final boolean shuttingDown,
+ final boolean closeFirst) throws Exception
+ {
+ try (ServerSocket listener = new ServerSocket(0);
+ Socket peer = new Socket("127.0.0.1", listener.getLocalPort());
+ Socket socket = listener.accept())
+ {
+ final SSLSocket secure = new TestSocket(input);
+ final Session session = new Session(socket, secure);
+ session.setProtocolVersion(version);
+ final DataServerHandler handler = handler(session, shuttingDown);
+ if (closeFirst)
+ {
+ session.close();
+ }
+ try
+ {
+ return errorLogRecordsOf(() -> {
+ new ServerReader(session, handler).run();
+ return null;
+ });
+ }
+ finally
+ {
+ session.close();
+ }
+ }
+ }
+
+ private static DataServerHandler handler(final Session session, final boolean shuttingDown) throws Exception
+ {
+ final DataServerHandler handler = mock(DataServerHandler.class);
+ final ReplicationServer server = mock(ReplicationServer.class);
+ when(server.getServerId()).thenReturn(9301);
+ final java.lang.reflect.Field field = MessageHandler.class.getDeclaredField("replicationServer");
+ field.setAccessible(true);
+ field.set(handler, server);
+ when(handler.shuttingDown()).thenReturn(shuttingDown);
+ when(handler.getBadlyDisconnectedErrorMessage()).thenReturn(ERR_DS_BADLY_DISCONNECTED.get(
+ 9301, 9311, session.getReadableRemoteAddress(), DN.valueOf(TEST_ROOT_DN_STRING)));
+ return handler;
+ }
+
+ private static final class TestSocket extends SSLSocket
+ {
+ private final InputStream input;
+ private final OutputStream output = new ByteArrayOutputStream();
+
+ TestSocket(final byte[] bytes) { this(new ByteArrayInputStream(bytes)); }
+ TestSocket(final InputStream input) { this.input = input; }
+ @Override public InputStream getInputStream() { return input; }
+ @Override public OutputStream getOutputStream() { return output; }
+ @Override public String[] getSupportedCipherSuites() { return new String[0]; }
+ @Override public String[] getEnabledCipherSuites() { return new String[0]; }
+ @Override public void setEnabledCipherSuites(String[] suites) { }
+ @Override public String[] getSupportedProtocols() { return new String[0]; }
+ @Override public String[] getEnabledProtocols() { return new String[0]; }
+ @Override public void setEnabledProtocols(String[] protocols) { }
+ @Override public javax.net.ssl.SSLSession getSession() { return null; }
+ @Override public void addHandshakeCompletedListener(javax.net.ssl.HandshakeCompletedListener listener) { }
+ @Override public void removeHandshakeCompletedListener(javax.net.ssl.HandshakeCompletedListener listener) { }
+ @Override public void startHandshake() throws IOException { }
+ @Override public void setUseClientMode(boolean mode) { }
+ @Override public boolean getUseClientMode() { return false; }
+ @Override public void setNeedClientAuth(boolean need) { }
+ @Override public boolean getNeedClientAuth() { return false; }
+ @Override public void setWantClientAuth(boolean want) { }
+ @Override public boolean getWantClientAuth() { return false; }
+ @Override public void setEnableSessionCreation(boolean flag) { }
+ @Override public boolean getEnableSessionCreation() { return false; }
+ }
+}
--
Gitblit v1.10.0