From 073f4eda43af479cd7645f5ab8a048932c612bcf Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 03 Nov 2026 12:24:53 +0000
Subject: [PATCH] [#1265] Avoid monitor request errors during local connection shutdown (#1266)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java | 13 +
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorRequestTest.java | 399 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 408 insertions(+), 4 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
index d92aab4..bf20eab 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationDomainMonitor.java
@@ -145,14 +145,19 @@
{
rs.send(msg);
- // Only register this server ID to pending table if we were able
- // to send the message.
+ // Register an expected response when send returns without throwing.
+ // A request dropped during concurrent session close is still awaited.
pendingMonitorDataServerIDs.add(serverId);
}
catch (IOException e)
{
- // Log a message and do a best effort from here.
- logger.error(ERR_SENDING_REMOTE_MONITOR_DATA_REQUEST, baseDN, serverId, e.getMessage());
+ logger.traceException(e);
+ // The reader closes its session before doStop() marks the handler as shutting down.
+ // Check after the send: local close or shutdown may begin after handler selection.
+ if (!rs.shuttingDown() && !rs.session.closeInitiated())
+ {
+ logger.error(ERR_SENDING_REMOTE_MONITOR_DATA_REQUEST, baseDN, serverId, e.getMessage());
+ }
}
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorRequestTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorRequestTest.java
new file mode 100644
index 0000000..6c27464
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationDomainMonitorRequestTest.java
@@ -0,0 +1,399 @@
+/*
+ * 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.mockito.Mockito.*;
+
+import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+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 java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.forgerock.opendj.ldap.DN;
+import org.mockito.invocation.InvocationOnMock;
+import org.opends.server.loggers.ConsoleDebugLogPublisher;
+import org.opends.server.loggers.DebugLogger;
+import org.opends.server.loggers.TraceSettings;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.protocol.MonitorRequestMsg;
+import org.opends.server.replication.protocol.ProtocolVersion;
+import org.opends.server.replication.protocol.ReplSessionSecurity;
+import org.opends.server.replication.protocol.Session;
+import org.opends.server.util.StaticUtils;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/** Monitor requests racing with local cleanup must not report ERROR 108 (#1265). */
+@SuppressWarnings("javadoc")
+public class ReplicationDomainMonitorRequestTest extends ReplicationTestCase
+{
+ private static final int TIMEOUT_MS = 10000;
+ private static final DN BASE_DN = DN.valueOf("dc=example,dc=com");
+
+ @DataProvider
+ public Object[][] failedSends()
+ {
+ return new Object[][] {
+ // Local close, handler shutdown, ERROR 108 records.
+ { true, false, 0 },
+ { true, true, 0 },
+ { false, false, 1 },
+ { false, true, 0 },
+ };
+ }
+
+ @Test(dataProvider = "failedSends")
+ public void failedRequestChecksLocalCloseAndHandlerShutdown(final boolean locallyClosed,
+ final boolean shuttingDown, final int expectedErrors) throws Exception
+ {
+ final Session[] pair = connectSessionPair();
+ try
+ {
+ final Session session = pair[1];
+ if (locallyClosed)
+ {
+ session.start();
+ session.waitForStartup();
+ session.close();
+ }
+ else
+ {
+ // An unstarted publisher sends synchronously. Break the socket without marking local close.
+ closeSocketsUnder(session);
+ }
+ assertThat(session.closeInitiated()).isEqualTo(locallyClosed);
+ final Requesting requesting = new Requesting(session, shuttingDown, 9302);
+ final List<String> logged = requesting.recompute();
+
+ requesting.assertFailedWithoutWaiting(logged);
+ assertRequestErrors(logged, expectedErrors);
+ if (expectedErrors != 0)
+ {
+ assertThat(logged).filteredOn(record -> record.contains("severity=ERROR msgID=108 "))
+ .allMatch(record -> record.contains(requesting.failure.get().getMessage()));
+ }
+ }
+ finally
+ {
+ StaticUtils.close(pair);
+ }
+ }
+
+ @Test
+ public void requestAfterConcurrentLocalCloseIsNotReported() throws Exception
+ {
+ final Session[] pair = connectSessionPair();
+ try
+ {
+ final Session session = pair[1];
+ session.start();
+ session.waitForStartup();
+ final Requesting requesting = new Requesting(session, false, 9302);
+ final CountDownLatch sending = new CountDownLatch(1);
+ final CountDownLatch closed = new CountDownLatch(1);
+ doAnswer(invocation ->
+ {
+ // Hold send after handler selection until the other thread completes local close.
+ sending.countDown();
+ assertThat(closed.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("local close completed").isTrue();
+ return requesting.send(invocation);
+ }).when(requesting.handler).send(any(MonitorRequestMsg.class));
+
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try
+ {
+ final Future<List<String>> sender = executor.submit(requesting::recompute);
+ assertThat(sending.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("handler selected").isTrue();
+ assertThat(session.closeInitiated()).isFalse();
+ session.close();
+ assertThat(session.closeInitiated()).isTrue();
+ closed.countDown();
+ final List<String> logged = sender.get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ requesting.assertFailedWithoutWaiting(logged);
+ assertRequestErrors(logged, 0);
+ }
+ finally
+ {
+ closed.countDown();
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("sender stopped").isTrue();
+ }
+ }
+ finally
+ {
+ StaticUtils.close(pair);
+ }
+ }
+
+ @Test
+ public void requestAfterHandlerShutdownAtSelectionIsNotReported() throws Exception
+ {
+ final Session[] pair = connectSessionPair();
+ try
+ {
+ final Session session = pair[1];
+ closeSocketsUnder(session);
+ final Requesting requesting = new Requesting(session, false, 9302);
+ doAnswer(invocation ->
+ {
+ // Model stopServer() engaging shutdown after selection, before the real send fails.
+ requesting.shutdown.set(true);
+ return requesting.send(invocation);
+ }).when(requesting.handler).send(any(MonitorRequestMsg.class));
+
+ final List<String> logged = requesting.recompute();
+ assertThat(session.closeInitiated()).isFalse();
+ requesting.assertFailedWithoutWaiting(logged);
+ assertRequestErrors(logged, 0);
+ }
+ finally
+ {
+ StaticUtils.close(pair);
+ }
+ }
+
+ @Test
+ public void suppressedRequestFailureIsStillTraced() throws Exception
+ {
+ final Session[] pair = connectSessionPair();
+ try
+ {
+ final Session session = pair[1];
+ session.start();
+ session.waitForStartup();
+ session.close();
+ final Requesting requesting = new Requesting(session, false, 9302);
+ final ByteArrayOutputStream trace = new ByteArrayOutputStream();
+ try (PrintStream output = new PrintStream(trace, true, "UTF-8"))
+ {
+ // Capture only recomputation, independently of global test trace settings.
+ final ConsoleDebugLogPublisher capture = new ConsoleDebugLogPublisher(output);
+ capture.addTraceSettings(ReplicationDomainMonitor.class.getName() + "#recomputeMonitorData",
+ new TraceSettings());
+ DebugLogger.getInstance().addLogPublisher(capture);
+ try
+ {
+ final List<String> logged = requesting.recompute();
+
+ requesting.assertFailedWithoutWaiting(logged);
+ assertRequestErrors(logged, 0);
+ assertThat(trace.toString("UTF-8"))
+ .contains("caught={" + requesting.failure.get() + "}", "recomputeMonitorData()");
+ }
+ finally
+ {
+ DebugLogger.getInstance().removeLogPublisher(capture);
+ }
+ }
+ }
+ finally
+ {
+ StaticUtils.close(pair);
+ }
+ }
+
+ @DataProvider
+ public Object[][] precedingFailure()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "precedingFailure")
+ public void deliveredRequestExpectsAResponseEvenAfterAnotherSendFails(final boolean precedingFailure)
+ throws Exception
+ {
+ final Session[] pair = connectSessionPair();
+ try
+ {
+ final Requesting requesting = new Requesting(pair[1], false, 9302);
+ // A separate closed session must not prevent the next peer from receiving its request.
+ final Session[] broken = connectSessionPair();
+ try
+ {
+ broken[1].close();
+ final Requesting failed = new Requesting(broken[1], false, 9303);
+ if (precedingFailure)
+ {
+ final Map<Integer, ReplicationServerHandler> peers = new LinkedHashMap<>();
+ peers.put(9303, failed.handler);
+ peers.put(9302, requesting.handler);
+ when(requesting.domain.getConnectedRSs()).thenReturn(peers);
+ }
+ final List<String> logged = requesting.recompute();
+
+ final MonitorRequestMsg received = (MonitorRequestMsg) pair[0].receive();
+ assertThat(received.getSenderID()).isEqualTo(9301);
+ assertThat(received.getDestination()).isEqualTo(9302);
+ assertThat(requesting.failure.get()).isNull();
+ assertRequestErrors(logged, 0);
+ // The peer deliberately stays silent: this delivered request must register an expected response.
+ assertThat(logged).filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID=106 "))
+ .hasSize(1).allMatch(record -> record.contains("RS(9302)"));
+ if (precedingFailure)
+ {
+ assertThat(failed.failure.get()).as("preceding real send failed").isNotNull();
+ }
+ }
+ finally
+ {
+ StaticUtils.close(broken);
+ }
+ }
+ finally
+ {
+ StaticUtils.close(pair);
+ }
+ }
+
+ private static void assertRequestErrors(final List<String> logged, final int count)
+ {
+ assertThat(logged).filteredOn(record -> record.contains("category=SYNC severity=ERROR msgID=108 "))
+ .hasSize(count).allMatch(record -> record.contains("domain \"dc=example,dc=com\" to replication server RS(9302)"));
+ }
+
+ /** Real monitor recomputation and handler send, without starting a replication topology. */
+ private static final class Requesting
+ {
+ private final ReplicationServerHandler handler = mock(ReplicationServerHandler.class);
+ private final ReplicationServerDomain domain = mock(ReplicationServerDomain.class);
+ private final ReplicationDomainMonitor monitor;
+ private final AtomicBoolean shutdown = new AtomicBoolean();
+ private final AtomicReference<IOException> failure = new AtomicReference<>();
+
+ private Requesting(final Session session, final boolean shuttingDown, final int peerId) throws Exception
+ {
+ setField(ServerHandler.class, handler, "session", session);
+ setField(MessageHandler.class, handler, "replicationServerDomain", domain);
+ shutdown.set(shuttingDown);
+ when(handler.shuttingDown()).thenAnswer(invocation -> shutdown.get());
+ when(handler.getServerId()).thenReturn(peerId);
+ doAnswer(invocation -> send(invocation)).when(handler).send(any(MonitorRequestMsg.class));
+ when(domain.getBaseDN()).thenReturn(BASE_DN);
+ when(domain.getLocalRSServerId()).thenReturn(9301);
+ when(domain.getConnectedDSs()).thenReturn(Collections.emptyMap());
+ when(domain.getConnectedRSs()).thenReturn(Collections.singletonMap(peerId, handler));
+ when(domain.getLatestServerState()).thenReturn(new ServerState());
+ monitor = new ReplicationDomainMonitor(domain);
+ }
+
+ private List<String> recompute() throws Exception
+ {
+ return errorLogRecordsOf(() ->
+ {
+ final ReplicationDomainMonitorData data = monitor.recomputeMonitorData();
+ assertThat(data).isSameAs(monitor.getMonitorData());
+ assertThat(data.getRSStates(9301)).as("best-effort local monitor data").isNotNull();
+ return null;
+ });
+ }
+
+ /** Retain the real send failure so skipping send cannot pass a suppression case. */
+ private Object send(final InvocationOnMock invocation) throws Throwable
+ {
+ try
+ {
+ return invocation.callRealMethod();
+ }
+ catch (IOException e)
+ {
+ failure.set(e);
+ throw e;
+ }
+ }
+
+ private void assertFailedWithoutWaiting(final List<String> logged)
+ {
+ assertThat(failure.get()).as("real send failure").isNotNull();
+ assertThat(logged).noneMatch(record -> record.contains("severity=WARNING msgID=106 "));
+ verify(domain, never()).stopServer(any(ServerHandler.class), anyBoolean());
+ }
+ }
+
+ private static void setField(final Class<?> owner, final Object target, final String name, final Object value)
+ throws Exception
+ {
+ final Field field = owner.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ /** Break the socket without calling Session.close(), keeping closeInitiated false. */
+ private static void closeSocketsUnder(final Session session) throws Exception
+ {
+ for (final String name : new String[] { "plainSocket", "secureSocket" })
+ {
+ final Field field = Session.class.getDeclaredField(name);
+ field.setAccessible(true);
+ StaticUtils.close((Closeable) field.get(session));
+ }
+ }
+
+ /** A real TLS session pair, client first, with bounded reads and neither publisher started. */
+ private static Session[] connectSessionPair() throws Exception
+ {
+ final ReplSessionSecurity security = getReplSessionSecurity();
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (ServerSocket listener = new ServerSocket(0))
+ {
+ final Socket clientSocket = new Socket("127.0.0.1", listener.getLocalPort());
+ Socket serverSocket = null;
+ Session serverEnd = null;
+ boolean connected = false;
+ try
+ {
+ final Future<Session> clientEnd = executor.submit(() -> security.createClientSession(clientSocket, TIMEOUT_MS));
+ serverSocket = listener.accept();
+ serverEnd = security.createServerSession(serverSocket, TIMEOUT_MS);
+ final Session[] pair = { clientEnd.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
+ for (final Session session : pair)
+ {
+ session.setProtocolVersion(ProtocolVersion.getCurrentVersion());
+ }
+ connected = true;
+ return pair;
+ }
+ finally
+ {
+ if (!connected)
+ {
+ // Closing the sockets also releases a client handshake still running in the executor.
+ StaticUtils.close(clientSocket, serverSocket, serverEnd);
+ }
+ }
+ }
+ finally
+ {
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("session setup stopped").isTrue();
+ }
+ }
+}
--
Gitblit v1.10.0