opendj-server-legacy/src/main/java/org/opends/server/protocols/jmx/DirectoryRMIServerSocketFactory.java

@@ -19,6 +19,7 @@

import org.forgerock.i18n.slf4j.LocalizedLogger;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
@@ -37,7 +38,7 @@
 * Sockets Layer (SSL) or Transport Layer Security (TLS) protocols.
 */
public class DirectoryRMIServerSocketFactory implements
    RMIServerSocketFactory
    RMIServerSocketFactory, Closeable
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

@@ -50,6 +51,9 @@
  /** The address to listen on, which could be INADDR_ANY. */
  private final InetAddress listenAddress;

  /** The server socket owned by this factory. */
  private ServerSocket serverSocket;

  /**
   * Constructs a new <code>DirectoryRMIServerSocketFactory</code> with the
   * specified SSL socket configuration.
@@ -105,7 +109,7 @@
  @Override
  public ServerSocket createServerSocket(int port) throws IOException
  {
    return new ServerSocket(port, 0, listenAddress)
    serverSocket = new ServerSocket(port, 0, listenAddress)
    {
      @Override
      public Socket accept() throws IOException
@@ -127,6 +131,22 @@
        return sslSocket;
      }
    };
    return serverSocket;
  }

  /**
   * Closes the underlying server socket, including an anonymous RMI listener.
   *
   * @throws IOException if the socket cannot be closed
   */
  @Override
  public void close() throws IOException
  {
    if (serverSocket != null)
    {
      serverSocket.close();
      serverSocket = null;
    }
  }

  /**

 opendj-server-legacy/src/main/java/org/opends/server/protocols/jmx/OpendsRmiServerSocketFactory.java

@@ -13,9 +13,11 @@
 *
 * Copyright 2006-2008 Sun Microsystems, Inc.
 * Portions Copyright 2014-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.protocols.jmx;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
@@ -38,7 +40,7 @@
 * <p>
 * Each connector is registered into the JMX MBean server.
 */
public class OpendsRmiServerSocketFactory implements RMIServerSocketFactory
public class OpendsRmiServerSocketFactory implements RMIServerSocketFactory, Closeable
{
  /** The address to listen on, which could be INADDR_ANY. */
  private final InetAddress listenAddress;
@@ -69,7 +71,8 @@
   *
   * @throws IOException If an I/O error occurs when closing the socket.
   */
  void close() throws IOException
  @Override
  public void close() throws IOException
  {
    if (serverSocket != null)
    {

 opendj-server-legacy/src/main/java/org/opends/server/protocols/jmx/RmiConnector.java

@@ -20,6 +20,7 @@
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.util.StaticUtils.*;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
@@ -165,6 +166,9 @@
   */
  private OpendsRmiServerSocketFactory rmiSsf;

  /** The socket factory owned by the RMI connector, separate from the registry's. */
  private Closeable connectorSocketFactory;

  /**
   * The RMI protocol version used by this connector.
   */
@@ -371,8 +375,10 @@
        SSLSocketFactory ssf = ctx.getSocketFactory();

        // set the Server socket factory in the JMX map
        rmiServerSockeyFactory = new DirectoryRMIServerSocketFactory(
        DirectoryRMIServerSocketFactory factory = new DirectoryRMIServerSocketFactory(
            ssf, false, jmxConnectionHandler.getListenAddress());
        connectorSocketFactory = factory;
        rmiServerSockeyFactory = factory;
        env.put(
            "jmx.remote.rmi.server.socket.factory",
            rmiServerSockeyFactory);
@@ -393,8 +399,10 @@
        {
          logger.trace("UNSECURE CONNECTION");
        }
        rmiServerSockeyFactory = new OpendsRmiServerSocketFactory(
        OpendsRmiServerSocketFactory factory = new OpendsRmiServerSocketFactory(
            jmxConnectionHandler.getListenAddress());
        connectorSocketFactory = factory;
        rmiServerSockeyFactory = factory;
      }

      // specify the rmi JMX authenticator to be used
@@ -553,6 +561,11 @@
    jmxRmiConnectorNoClientCertificate = null;
    jmxRmiConnectorClientCertificate = null;

    // RMI keeps anonymous-port listeners open even after the last object is unexported.
    // Close our socket explicitly, also when stop() failed or the registry is being kept.
    close(connectorSocketFactory);
    connectorSocketFactory = null;

    // Unregister connectors and stop them.
    try
    {

 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/main/java/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -2969,12 +2969,15 @@
        }
        catch (IOException e)
        {
          // We log the error. The requestor will detect a timeout or
          // any other failure on the connection.

          // The requestor will detect a timeout or any other failure on the connection.
          // FIXME: why do we log for RSs but not DSs?
          logger.traceException(e);
          logger.error(ERR_CHANGELOG_ERROR_SENDING_MSG, msg.getValue().getDestination());
          // 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_CHANGELOG_ERROR_SENDING_MSG, msg.getValue().getDestination());
          }
        }
      }
    }

 opendj-server-legacy/src/test/java/org/opends/server/protocols/jmx/JmxListenAddressTestCase.java

@@ -21,6 +21,7 @@

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.rmi.registry.LocateRegistry;
@@ -28,6 +29,8 @@
import java.security.cert.X509Certificate;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.management.remote.JMXConnector;
import javax.net.ssl.SSLContext;
@@ -201,6 +204,83 @@
    }
  }

  /** Each handler restart must release the anonymous RMI port, with and without SSL. */
  @Test(dataProvider = "useSSL")
  public void releasesTheAnonymousRmiPortAfterEachStop(boolean useSSL) throws Exception
  {
    final int listenPort = TestCaseUtils.findFreePort();
    for (int i = 0; i < 3; i++)
    {
      JmxConnectionHandler handler = newHandler(listenAddress, listenPort, 0, useSSL);
      final int rmiPort;
      try
      {
        handler.run();
        rmiPort = anonymousRmiPort(listenAddress, listenPort);
        assertOpensASession(listenAddress, listenPort);
      }
      finally
      {
        handler.finalizeConnectionHandler(STOP_REASON);
        // Shutdown may be called both by processServerShutdown and by finalization.
        handler.finalizeConnectionHandler(STOP_REASON);
      }
      assertPortReleased(listenAddress, rmiPort);
      assertPortReleased(listenAddress, listenPort);
    }
  }

  /** Restarting only the connector must close its old socket while keeping the registry usable. */
  @Test(dataProvider = "useSSL")
  public void releasesTheAnonymousRmiPortAfterAConfigurationChange(boolean useSSL) throws Exception
  {
    final int listenPort = TestCaseUtils.findFreePort();
    JmxConnectionHandler handler = newHandler(listenAddress, listenPort, 0, useSSL);
    try
    {
      handler.run();
      final int oldRmiPort = anonymousRmiPort(listenAddress, listenPort);
      assertOpensASession(listenAddress, listenPort);

      // Select the replacement port while the old connector is listening: it cannot be the old port.
      final int newRmiPort = TestCaseUtils.findFreePort();
      applyConfigurationChange(handler, newConfig(listenAddress, listenPort, newRmiPort, useSSL));

      assertPortReleased(listenAddress, oldRmiPort);
      assertOpensASession(listenAddress, listenPort);
    }
    finally
    {
      handler.finalizeConnectionHandler(STOP_REASON);
    }
  }

  /**
   * The configured port is zero, so read the assigned port from the stub's endpoint description.
   * This avoids opening JDK-internal RMI packages for reflection in the test JVM.
   */
  private static int anonymousRmiPort(InetAddress address, int listenPort) throws Exception
  {
    Registry registry = LocateRegistry.getRegistry(address.getHostAddress(), listenPort);
    String stub = registry.lookup("org.opends.server.protocols.jmx.client-unknown").toString();
    Matcher endpoint = Pattern.compile("endpoint:\\[[^\\]]*?:(\\d+)[,\\]]").matcher(stub);
    assertTrue(endpoint.find(), "Cannot find the RMI endpoint in " + stub);
    int port = Integer.parseInt(endpoint.group(1));
    assertTrue(port > 0, "The RMI stub must advertise the assigned port");
    assertTrue(isAcceptingConnections(address, port), "The RMI connector must listen before it is stopped");
    return port;
  }

  private static void assertPortReleased(InetAddress address, int port) throws IOException
  {
    assertFalse(isAcceptingConnections(address, port), "The stopped RMI listener still accepts connections on " + port);
    try (ServerSocket socket = new ServerSocket())
    {
      socket.setReuseAddress(true);
      socket.bind(new InetSocketAddress(address, port));
    }
  }

  /** A change that restarts the RMI connector keeps the listen address that is reported. */
  @Test
  public void reportsTheListenAddressAfterAChangeOfTheRmiPort() throws Exception

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

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

New file
@@ -0,0 +1,389 @@
/*
 * 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.Constructor;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collections;
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 java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

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.MonitorMsg;
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 replies racing with a local close must not report an unexpected replication failure (#1262). */
@SuppressWarnings("javadoc")
public class ReplicationServerDomainMonitorTest extends ReplicationTestCase
{
  private static final int TIMEOUT_MS = 10000;

  @DataProvider
  public Object[][] failedSends()
  {
    return new Object[][] {
      // Local close, handler shutdown, ERROR 66 records.
      { true, false, 0 },
      { true, true, 0 },
      { false, false, 1 },
      { false, true, 0 },
    };
  }

  @Test(dataProvider = "failedSends")
  public void failedMonitorChecksLocalCloseAndHandlerShutdown(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
      {
        /*
         * Leave the publisher unstarted so the real send fails synchronously, without marking
         * a local close. A running publisher handles its own write failures instead of throwing
         * them back to the monitor sender. This pins the diagnostic for an otherwise open
         * handler, independently of the close flag.
         */
        closeSocketsUnder(session);
      }
      assertThat(session.closeInitiated()).isEqualTo(locallyClosed);
      final Forwarding forwarding = new Forwarding(session, shuttingDown);

      final List<String> logged = errorLogRecordsOf(() ->
      {
        forwarding.domain.sendPendingStatusMessages();
        return null;
      });

      verify(forwarding.handler).send(forwarding.monitor);
      forwarding.assertFailedWithoutStopping();
      assertMonitorErrors(logged, expectedErrors);
    }
    finally
    {
      StaticUtils.close(pair);
    }
  }

  @Test
  public void monitorAfterConcurrentLocalCloseIsNotReported() throws Exception
  {
    final Session[] pair = connectSessionPair();
    try
    {
      final Session session = pair[1];
      session.start();
      session.waitForStartup();
      final Forwarding forwarding = new Forwarding(session, false);
      final CountDownLatch sending = new CountDownLatch(1);
      final CountDownLatch closed = new CountDownLatch(1);
      doAnswer(invocation ->
      {
        // The domain has selected this handler and entered send while its session is still open.
        sending.countDown();
        assertThat(closed.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("local close completed").isTrue();
        return forwarding.send(invocation);
      }).when(forwarding.handler).send(forwarding.monitor);

      final List<String> logged = errorLogRecordsOf(() ->
      {
        final ExecutorService executor = Executors.newSingleThreadExecutor();
        try
        {
          final Future<?> sender = executor.submit(forwarding.domain::sendPendingStatusMessages);
          assertThat(sending.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)).as("handler selected").isTrue();
          assertThat(session.closeInitiated()).isFalse();
          session.close();
          assertThat(session.closeInitiated()).isTrue();
          closed.countDown();
          sender.get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
        }
        finally
        {
          closed.countDown();
          executor.shutdownNow();
          assertThat(executor.awaitTermination(TIMEOUT_MS, TimeUnit.MILLISECONDS))
              .as("sender stopped before log capture is removed").isTrue();
        }
        return null;
      });

      verify(forwarding.handler).send(forwarding.monitor);
      forwarding.assertFailedWithoutStopping();
      assertMonitorErrors(logged, 0);
    }
    finally
    {
      StaticUtils.close(pair);
    }
  }

  @Test
  public void monitorAfterConcurrentHandlerShutdownIsNotReported() throws Exception
  {
    final Session[] pair = connectSessionPair();
    try
    {
      final Session session = pair[1];
      closeSocketsUnder(session);
      final Forwarding forwarding = new Forwarding(session, false);
      doAnswer(invocation ->
      {
        // Model another stopServer() engaging shutdown after the domain selected this handler.
        forwarding.shutdown.set(true);
        return forwarding.send(invocation);
      }).when(forwarding.handler).send(forwarding.monitor);

      final List<String> logged = errorLogRecordsOf(() ->
      {
        forwarding.domain.sendPendingStatusMessages();
        return null;
      });

      assertThat(session.closeInitiated()).isFalse();
      verify(forwarding.handler).send(forwarding.monitor);
      forwarding.assertFailedWithoutStopping();
      assertMonitorErrors(logged, 0);
    }
    finally
    {
      StaticUtils.close(pair);
    }
  }

  @Test
  public void suppressedMonitorFailureIsStillTraced() throws Exception
  {
    final Session[] pair = connectSessionPair();
    try
    {
      final Session session = pair[1];
      session.start();
      session.waitForStartup();
      session.close();
      final Forwarding forwarding = new Forwarding(session, false);

      final ByteArrayOutputStream trace = new ByteArrayOutputStream();
      try (PrintStream output = new PrintStream(trace, true, "UTF-8"))
      {
        // Capture only this forwarding method, independently of global test trace settings.
        final ConsoleDebugLogPublisher capture = new ConsoleDebugLogPublisher(output);
        capture.addTraceSettings(ReplicationServerDomain.class.getName() + "#sendPendingMonitorMsgs",
            new TraceSettings());
        DebugLogger.getInstance().addLogPublisher(capture);
        try
        {
          final List<String> logged = errorLogRecordsOf(() ->
          {
            forwarding.domain.sendPendingStatusMessages();
            return null;
          });

          forwarding.assertFailedWithoutStopping();
          assertMonitorErrors(logged, 0);
          assertThat(trace.toString("UTF-8"))
              .contains("caught={" + forwarding.failure.get() + "}", "sendPendingMonitorMsgs()");
        }
        finally
        {
          DebugLogger.getInstance().removeLogPublisher(capture);
        }
      }
    }
    finally
    {
      StaticUtils.close(pair);
    }
  }

  @Test
  public void monitorIsDeliveredToAnOpenSession() throws Exception
  {
    final Session[] pair = connectSessionPair();
    try
    {
      final Forwarding forwarding = new Forwarding(pair[1], false);
      final List<String> logged = errorLogRecordsOf(() ->
      {
        forwarding.domain.sendPendingStatusMessages();
        return null;
      });

      final MonitorMsg received = (MonitorMsg) pair[0].receive();
      assertThat(received.getSenderID()).isEqualTo(9301);
      assertThat(received.getDestination()).isEqualTo(9302);
      assertThat(received.getReplServerDbState().isEmpty()).isTrue();
      assertThat(forwarding.failure.get()).isNull();
      verify(forwarding.domain, never()).stopServer(any(ServerHandler.class), anyBoolean());
      assertMonitorErrors(logged, 0);
    }
    finally
    {
      StaticUtils.close(pair);
    }
  }

  private static void assertMonitorErrors(final List<String> logged, final int count)
  {
    // Keep duplicates: two identical reports in the same second must count as two errors.
    assertThat(logged).filteredOn(record -> record.contains("category=SYNC severity=ERROR msgID=66 "))
        .hasSize(count)
        .allMatch(record -> record.contains("sending a Message to 9302."));
  }

  /** Real status-message forwarding and handler send, without a replication topology or teardown. */
  private static final class Forwarding
  {
    private final MonitorMsg monitor = new MonitorMsg(9301, 9302);
    private final ReplicationServerHandler handler = mock(ReplicationServerHandler.class);
    private final ReplicationServerDomain domain = mock(ReplicationServerDomain.class);
    private final AtomicBoolean shutdown = new AtomicBoolean();
    private final AtomicReference<IOException> failure = new AtomicReference<>();

    private Forwarding(final Session session, final boolean shuttingDown) throws Exception
    {
      monitor.setReplServerDbState(new ServerState());
      setField(ServerHandler.class, handler, "session", session);
      setField(MessageHandler.class, handler, "replicationServerDomain", domain);
      shutdown.set(shuttingDown);
      when(handler.shuttingDown()).thenAnswer(invocation -> shutdown.get());
      doAnswer(invocation -> send(invocation)).when(handler).send(monitor);

      final Class<?> pendingClass = Class.forName(ReplicationServerDomain.class.getName() + "$PendingStatusMessages");
      final Constructor<?> constructor = pendingClass.getDeclaredConstructor();
      constructor.setAccessible(true);
      final Object pending = constructor.newInstance();
      setField(pendingClass, pending, "pendingRSMonitorMsgs", Collections.singletonMap(9302, monitor));
      setField(ReplicationServerDomain.class, domain, "pendingStatusMessages", pending);
      setField(ReplicationServerDomain.class, domain, "pendingStatusMessagesLock", new Object());
      setField(ReplicationServerDomain.class, domain, "connectedRSs", Collections.singletonMap(9302, handler));
      doCallRealMethod().when(domain).sendPendingStatusMessages();
    }

    /** Call the real send and retain its IOException so a skipped 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 assertFailedWithoutStopping()
    {
      assertThat(failure.get()).as("real send failure").isNotNull();
      // Monitor replies leave failure detection and cleanup to the requester/reader.
      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();
    }
  }
}