[#1265] Avoid monitor request errors during local connection shutdown (#1266)

Valery Kharseko

10 hours ago 073f4eda43af479cd7645f5ab8a048932c612bcf

[#1265] Avoid monitor request errors during local connection shutdown (#1266)

Fixes #1265

A monitor request can retain an RS handler whose session the reader has
already closed. The real send then fails on the closed socket, and
monitor recomputation previously logged ERROR 108 even during local
cleanup.

`ReplicationDomainMonitor` now reports ERROR 108 only when the handler
is not shutting down and local session close has not begun. Both flags
are checked after the failed send. Exception tracing is unconditional,
including when ERROR 108 is suppressed. Failures on otherwise live
handlers retain the existing diagnostic. Best-effort recomputation
continues to other peers. Only requests whose send returns without
throwing register an expected response. A request dropped during
concurrent session close may still be awaited and produce WARN 106; this
existing pending-response behavior is unchanged. Convergence error
filters remain unchanged.

Nine regression cases use real TLS sessions and the real handler send
path, with domain construction and handler lifecycle state controlled by
the fixture. They cover all four close/shutdown combinations, local
close after handler selection, shutdown after selection, successful
delivery, continued delivery to a healthy peer after another send fails,
and exception tracing when ERROR 108 is suppressed. Failed-send cases
require an actual IOException and no missing-response warning; requests
delivered to a deliberately silent peer must produce that warning. The
trace case captures the real send's IOException through a method-scoped
debug publisher, removed in finally without replacing existing
publishers. The concurrent-close case releases the send after close()
completes; it does not cover a request dropped while close() is still in
progress.

## Validation

On JDK 21, based on master `cdc884c34f`:

- Without the trace call, only `suppressedRequestFailureIsStillTraced`
fails among the nine request cases: the real send throws
`SocketException: Socket closed`, but the captured trace is empty.
- Maven reactor `-Pprecommit verify`, including Javadoc, passes all 56
selected tests with no failures, errors or skips: 9 request, 3
monitor-state, 8 reply, 8 heartbeat, 8 reader, 4 writer, 6 session and
10 publisher-drain cases.
- Moving the trace call inside the ERROR guard fails only the new trace
case (1 failure out of 9). The same direct TestNG runner passes 9/9
against the unmodified build. The mutant was compiled outside reactor
build outputs.
- `git diff --check` passes. Independent local read-only review found no
issues.

Docker convergence and the full repository test suite were not run
locally.

```sh
mvn -o -Pprecommit verify -pl opendj-server-legacy -am \
  -Dit.test=ReplicationDomainMonitorRequestTest,ReplicationDomainMonitorTest,ReplicationServerDomainMonitorTest,ReplicationServerDomainHeartbeatTest,ServerReaderDisconnectTest,ServerWriterDisconnectTest,SessionTest,SessionPublisherDrainTest \
  -Dfailsafe.failIfNoSpecifiedTests=false \
  -Dtest=NoTestsSelected -Dsurefire.failIfNoSpecifiedTests=false
```

 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());
                }
              }
            }


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

New file
@@ -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();
    }
  }
}