[#792] Fail fast when the replication server cannot bind its listen port (#...

Valery Kharseko

22 hours ago 5380e8e895c410575fb9e07202992d31ed7cd8af

[#792] Fail fast when the replication server cannot bind its listen port (#795)

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplicationServerListener.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2008 Sun Microsystems, Inc.
 * Portions Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

@@ -56,15 +57,28 @@
      ReplicationSynchronizationProviderCfg configuration,
      DSRSShutdownSync dsrsShutdownSync) throws ConfigException
  {
    configuration.addReplicationServerAddListener(this);
    configuration.addReplicationServerDeleteListener(this);

    this.dsrsShutdownSync = dsrsShutdownSync;
    if (configuration.hasReplicationServer())
    {
      final ReplicationServerCfg cfg = configuration.getReplicationServer();
      // Registered only once the replication server exists: the listeners are held by the
      // server wide configuration repository and keyed by DN, so a listener registered by
      // a listener which is then discarded would be leaked until the server is restarted.
      replicationServer = new ReplicationServer(cfg, dsrsShutdownSync);
    }

    try
    {
      configuration.addReplicationServerAddListener(this);
      configuration.addReplicationServerDeleteListener(this);
    }
    catch (ConfigException e)
    {
      // This listener is not returned to MultimasterReplication, which is what stops the
      // replication server: nothing would ever shut down the one just created.
      shutdown();
      throw e;
    }
  }

  /** {@inheritDoc} */
@@ -78,9 +92,11 @@
    }
    catch (ConfigException e)
    {
      // we should never get to this point because the configEntry has
      // already been validated in configAddisAcceptable
      // The configEntry has already been validated in configAddisAcceptable, but the
      // listen port may have been taken since then: report why the creation failed. The
      // message is logged by the configuration handler, which reports the result.
      ccr.setResultCode(ResultCode.CONSTRAINT_VIOLATION);
      ccr.addMessage(e.getMessageObject());
    }
    return ccr;
  }

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java

@@ -39,8 +39,10 @@
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizableMessageBuilder;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.config.server.ConfigException;
@@ -92,7 +94,18 @@
{
  private String serverURL;

  private ServerSocket listenSocket;
  /**
   * Number of attempts to bind the listen port before giving up. The port may be held for a
   * short while by a socket which is being closed, so a few retries make the start-up
   * resilient to such transient conditions.
   */
  private static final int LISTEN_BIND_ATTEMPTS = 5;
  /** Delay between two attempts to bind the listen port. */
  private static final long LISTEN_BIND_RETRY_DELAY_MS = 200;
  /** Timeout of the diagnostic probe performed when the listen port cannot be bound. */
  private static final int LISTEN_PORT_PROBE_TIMEOUT_MS = 200;

  private volatile ServerSocket listenSocket;
  private Thread listenThread;
  private Thread connectThread;

@@ -109,8 +122,16 @@
  /** The backend that allow to search the changes (external changelog). */
  private ChangelogBackend changelogBackend;

  /**
   * Whether this instance registered the virtual attribute rules of the external changelog.
   * They are registered globally, by attribute name, so an instance which did not register
   * them must not deregister them: it would strip them from the instance which did.
   */
  private boolean externalChangelogRegistered;

  private final AtomicBoolean shutdown = new AtomicBoolean();
  private boolean stopListen;
  /** Written by the thread applying a configuration change, read by the listen thread. */
  private volatile boolean stopListen;
  private final ReplSessionSecurity replSessionSecurity;

  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
@@ -124,6 +145,16 @@
   */
  private static final Set<Integer> localPorts = new CopyOnWriteArraySet<>();

  /**
   * Number of attempts to bind a listen port which failed in this VM.
   * <p>
   * This is required for unit testing: it lets a test which holds a listen port release it
   * as soon as a replication server has actually failed to bind it, instead of after a
   * delay which would make the test vacuous when it is too long, and flaky when it is too
   * short.
   */
  static final AtomicInteger listenPortBindFailures = new AtomicInteger();

  /** Monitors for synchronizing domain creation with the connect thread. */
  private final Object domainTicketLock = new Object();
  private final Object connectThreadLock = new Object();
@@ -177,15 +208,25 @@
    this.dsrsShutdownSync = dsrsShutdownSync;
    this.domainPredicate = predicate;

    enableExternalChangeLog();
    ServerContext serverContext = DirectoryServer.getInstance().getServerContext();
    cryptoSuite = serverContext.getCryptoManager().
        newCryptoSuite(cfg.getCipherTransformation(), cfg.getCipherKeyLength(), cfg.isConfidentialityEnabled());
    try
    {
      enableExternalChangeLog();
      ServerContext serverContext = DirectoryServer.getInstance().getServerContext();
      cryptoSuite = serverContext.getCryptoManager().
          newCryptoSuite(cfg.getCipherTransformation(), cfg.getCipherKeyLength(), cfg.isConfidentialityEnabled());

    this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite);
      this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite);

    replSessionSecurity = new ReplSessionSecurity();
    initialize();
      replSessionSecurity = new ReplSessionSecurity();
      initialize();
    }
    catch (ConfigException e)
    {
      // This instance is never returned to the caller, so nothing will ever shut it down:
      // release what the initialization managed to acquire before failing.
      abortInitialization();
      throw e;
    }
    cfg.addChangeListener(this);

    localPorts.add(getReplicationPort());
@@ -441,8 +482,14 @@
    return true;
  }

  /** Initialization function for the replicationServer. */
  private void initialize()
  /**
   * Initialization function for the replicationServer.
   *
   * @throws ConfigException
   *           when the replication server cannot be started, in particular when its listen
   *           port cannot be bound.
   */
  private void initialize() throws ConfigException
  {
    shutdown.set(false);

@@ -451,8 +498,8 @@
      this.changelogDB.initializeDB();

      setServerURL();
      listenSocket = new ServerSocket();
      listenSocket.bind(new InetSocketAddress(getReplicationPort()));
      // Assigned before the threads are created, so that a failure below still releases it.
      listenSocket = bindListenPort(getReplicationPort());

      // creates working threads: we must first connect, then start to listen.
      if (logger.isTraceEnabled())
@@ -467,8 +514,7 @@
        logger.trace("RS " + getMonitorInstanceName() + " creates listen thread");
      }

      listenThread = new ReplicationServerListenThread(this);
      listenThread.start();
      startListenThread(listenSocket);

      if (logger.isTraceEnabled())
      {
@@ -476,14 +522,265 @@
      }
    } catch (UnknownHostException e)
    {
      logger.error(ERR_UNKNOWN_HOSTNAME);
      // Not logged here: the caller reports the ConfigException, logging it once.
      logger.traceException(e);
      throw new ConfigException(ERR_UNKNOWN_HOSTNAME.get(), e);
    } catch (IOException e)
    {
      logger.error(ERR_COULD_NOT_BIND_CHANGELOG, getReplicationPort(), e.getMessage());
      // A replication server whose listen port is not bound is dead: every consumer would
      // otherwise only learn about it as a "connection refused" somewhere else.
      logger.traceException(e);
      throw new ConfigException(bindFailureMessage(getReplicationPort(), e), e);
    }
  }

  /**
   * Binds the given port, retrying a few times when it is momentarily unavailable.
   *
   * @param port
   *          the port to bind
   * @return the bound listen socket
   * @throws IOException
   *           if the port could not be bound within {@link #LISTEN_BIND_ATTEMPTS} attempts
   */
  private ServerSocket bindListenPort(int port) throws IOException
  {
    for (int attempt = 1; ; attempt++)
    {
      final ServerSocket socket = new ServerSocket();
      try
      {
        socket.bind(new InetSocketAddress(port));
        if (attempt > 1)
        {
          logger.info(NOTE_BOUND_CHANGELOG_AFTER_RETRY, port, attempt);
        }
        return socket;
      }
      catch (IOException e)
      {
        close(socket);
        listenPortBindFailures.incrementAndGet();
        if (attempt >= LISTEN_BIND_ATTEMPTS)
        {
          throw e;
        }
        // The port is probed only when giving up: probing it on every attempt would cost a
        // connection to whoever holds it, and delay the next attempt for nothing.
        logger.warn(WARN_RETRYING_BIND_CHANGELOG, port, getExceptionMessage(e), LISTEN_BIND_RETRY_DELAY_MS);
        try
        {
          Thread.sleep(LISTEN_BIND_RETRY_DELAY_MS);
        }
        catch (InterruptedException e2)
        {
          Thread.currentThread().interrupt();
          throw e;
        }
      }
    }
  }

  /**
   * Stops the listen thread and releases the listen port.
   *
   * @throws InterruptedException
   *           if this thread is interrupted while waiting for the listen thread to stop
   */
  private void stopListenThread() throws InterruptedException
  {
    stopListen = true;
    close(listenSocket);
    if (listenThread != null)
    {
      listenThread.join();
      listenThread = null;
    }
  }

  /**
   * Starts a listen thread on the provided listen socket.
   * <p>
   * {@code stopListen} is only cleared here, i.e. once the listen port is bound, so a
   * failure to bind leaves this replication server consistently stopped rather than with a
   * listen thread which would spin on a closed socket.
   *
   * @param boundListenSocket
   *          the bound socket the listen thread will accept connections on
   */
  private void startListenThread(ServerSocket boundListenSocket)
  {
    listenSocket = boundListenSocket;
    stopListen = false;
    listenThread = new ReplicationServerListenThread(this);
    listenThread.start();
  }

  /**
   * Switches the listen port to the one of the provided configuration.
   * <p>
   * The new port is bound while the current one is still open and serving, so a failure
   * leaves this replication server listening on its current port, with its current
   * configuration: there is nothing to roll back, and no window during which this
   * replication server advertises a port that nothing listens to.
   *
   * @param newConfig
   *          the configuration being applied, whose listen port differs from the current one
   * @param ccr
   *          the result of the configuration change, to which a failure is added
   * @return {@code true} when this replication server listens on the new port, in which
   *         case {@code newConfig} has become its configuration
   */
  private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeResult ccr)
  {
    final ReplicationServerCfg previousConfig = this.config;
    final String previousServerURL = serverURL;
    final int newPort = newConfig.getReplicationPort();
    ServerSocket newListenSocket = null;
    try
    {
      // The current listen socket is still open and serving while the new port is bound.
      newListenSocket = bindListenPort(newPort);

      this.config = newConfig;
      setServerURL();

      stopListenThread();
      startListenThread(newListenSocket);
      newListenSocket = null;

      localPorts.remove(previousConfig.getReplicationPort());
      localPorts.add(newPort);
      return true;
    }
    catch (UnknownHostException e)
    {
      logger.traceException(e);
      ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
      ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get());
    }
    catch (IOException e)
    {
      // The new port could not be bound, the current listen socket was left untouched.
      logger.traceException(e);
      ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
      ccr.addMessage(bindFailureMessage(newPort, e));
    }
    catch (InterruptedException e)
    {
      // The previous listen thread may still be running, so do not hand it a new socket:
      // stopListen is still set, which makes that thread stop as soon as it wakes up.
      Thread.currentThread().interrupt();
      logger.traceException(e);
      ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
      ccr.addMessage(ERR_COULD_NOT_STOP_LISTEN_THREAD.get(getExceptionMessage(e)));
    }
    // The failure is reported through the ConfigChangeResult, which the configuration
    // handler logs: nothing of the new configuration was applied.
    this.config = previousConfig;
    serverURL = previousServerURL;
    close(newListenSocket);
    return false;
  }

  /**
   * Returns the message of a failed bind of the listen port, i.e. the error itself plus a
   * best effort diagnostic of what holds the port.
   *
   * @param port
   *          the port which could not be bound
   * @param cause
   *          the error returned by the failed bind
   * @return the message describing the failure
   */
  private static LocalizableMessage bindFailureMessage(int port, IOException cause)
  {
    return new LocalizableMessageBuilder(ERR_COULD_NOT_BIND_CHANGELOG.get(port, getExceptionMessage(cause)))
        .append(" ").append(describeListenPortHolder(port)).toMessage();
  }

  /**
   * Returns a best effort diagnostic of what holds the given port.
   * <p>
   * The port is probed over the loopback interface only, so the diagnostic reports what was
   * observed there and nothing more: a socket bound to another address holds the port
   * without ever accepting a loopback connection, and cannot be told apart from a socket
   * which does not accept connections at all, e.g. a client socket which was given that
   * port as its local port.
   *
   * @param port
   *          the port which could not be bound
   * @return the message describing what was observed on the loopback interface
   */
  private static LocalizableMessage describeListenPortHolder(int port)
  {
    final String probedAddress = InetAddress.getLoopbackAddress().getHostAddress() + ":" + port;
    try (Socket probe = new Socket())
    {
      // Without SO_REUSEADDR a probe which self-connects would, once closed, hold the port
      // in TIME_WAIT and defeat the very bind it is diagnosing.
      probe.setReuseAddress(true);
      probe.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), LISTEN_PORT_PROBE_TIMEOUT_MS);
      if (isSelfConnection(probe))
      {
        // The kernel can only give the probed port to the probe as its local port while
        // that port is free: whoever held it released it in the meantime.
        return ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE.get(probedAddress);
      }
      return ERR_COULD_NOT_BIND_CHANGELOG_LISTENING.get(probedAddress);
    }
    catch (IOException e)
    {
      logger.traceException(e);
      return ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING.get(probedAddress);
    }
  }

  /**
   * Releases what the initialization acquired before it failed.
   * <p>
   * It runs on a partially constructed instance, so every field it uses may still be
   * unassigned.
   */
  private void abortInitialization()
  {
    shutdown.set(true);
    if (connectThread != null)
    {
      connectThread.interrupt();
    }
    close(listenSocket);
    if (listenThread != null)
    {
      listenThread.interrupt();
    }
    shutdownExternalChangelog();
    if (this.changelogDB != null)
    {
      try
      {
        this.changelogDB.shutdownDB();
      }
      catch (ChangelogException ignored)
      {
        logger.traceException(ignored);
      }
    }
  }

  /**
   * Indicates whether this replication server has bound its listen port, i.e. whether it
   * can accept connections from directory servers and from other replication servers.
   *
   * @return {@code true} if this replication server is listening
   */
  public boolean isListening()
  {
    final ServerSocket socket = listenSocket;
    return socket != null && socket.isBound() && !socket.isClosed();
  }

  /**
   * Enable the external changelog if it is not already enabled.
   * <p>
   * The external changelog is provided by the changelog backend.
@@ -515,6 +812,9 @@
            getExceptionMessage(e)));
      }

      // Set before the rules are registered, so that a partial registration is released
      // too: this instance is then the one which registered whatever is registered.
      externalChangelogRegistered = true;
      registerVirtualAttributeRules();
    }
    catch (Exception e)
@@ -535,7 +835,14 @@
      changelogBackend.finalizeBackend();
      changelogBackend = null;
    }
    deregisterVirtualAttributeRules();
    if (externalChangelogRegistered)
    {
      // Virtual attribute rules are registered globally, by attribute name: deregistering
      // rules which this instance did not register would strip them from the instance
      // which did, e.g. when this one took the early return of enableExternalChangeLog().
      externalChangelogRegistered = false;
      deregisterVirtualAttributeRules();
    }
  }

  private List<VirtualAttributeRule> getVirtualAttributesRules() throws DirectoryException
@@ -863,17 +1170,14 @@
  {
    int port = configuration.getReplicationPort();

    try
    try (ServerSocket tmpSocket = new ServerSocket())
    {
      ServerSocket tmpSocket = new ServerSocket();
      tmpSocket.bind(new InetSocketAddress(port));
      tmpSocket.close();
      return true;
    }
    catch (Exception e)
    {
      LocalizableMessage message = ERR_COULD_NOT_BIND_CHANGELOG.get(port, e.getMessage());
      unacceptableReasons.add(message);
      unacceptableReasons.add(ERR_COULD_NOT_BIND_CHANGELOG.get(port, getExceptionMessage(e)));
      return false;
    }
  }
@@ -884,11 +1188,21 @@
  {
    final ConfigChangeResult ccr = new ConfigChangeResult();

    final Set<HostPort> oldRSAddresses = getConfiguredRSAddresses();
    final ReplicationServerCfg oldConfig = this.config;

    // Changing the listen port requires to stop the listen thread and restart it. It is
    // done first, and the new port is bound before the current one is released, so that a
    // change which cannot be applied leaves this replication server as it was, instead of
    // half configured and, worse, without any listener.
    if (configuration.getReplicationPort() != oldConfig.getReplicationPort()
        && !switchListenPort(configuration, ccr))
    {
      return ccr;
    }

    // Some of those properties change don't need specific code.
    // They will be applied for next connections. Some others have immediate effect
    final Set<HostPort> oldRSAddresses = getConfiguredRSAddresses();

    final ReplicationServerCfg oldConfig = this.config;
    this.config = configuration;

    disconnectRemovedReplicationServers(oldRSAddresses);
@@ -915,39 +1229,6 @@
    cryptoSuite.newParameters(config.getCipherTransformation(), config.getCipherKeyLength(),
        config.isConfidentialityEnabled());

    // changing the listen port requires to stop the listen thread
    // and restart it.
    if (getReplicationPort() != oldConfig.getReplicationPort())
    {
      stopListen = true;
      try
      {
        close(listenSocket);
        if (listenThread != null)
        {
          listenThread.join();
        }
        stopListen = false;

        setServerURL();
        listenSocket = new ServerSocket();
        listenSocket.bind(new InetSocketAddress(getReplicationPort()));

        listenThread = new ReplicationServerListenThread(this);
        listenThread.start();
      }
      catch (IOException e)
      {
        logger.traceException(e);
        logger.error(ERR_COULD_NOT_CLOSE_THE_SOCKET, e);
      }
      catch (InterruptedException e)
      {
        logger.traceException(e);
        logger.error(ERR_COULD_NOT_STOP_LISTEN_THREAD, e);
      }
    }

    // Update period value for monitoring publishers
    if (oldConfig.getMonitoringPeriod() != config.getMonitoringPeriod())
    {
@@ -1031,7 +1312,13 @@
  public boolean isConfigurationChangeAcceptable(
      ReplicationServerCfg configuration, List<LocalizableMessage> unacceptableReasons)
  {
    return true;
    if (configuration.getReplicationPort() == getReplicationPort())
    {
      return true;
    }
    // The change is persisted before it is applied, so rejecting a port which cannot be
    // bound is the only way to keep the configuration and the listen port in sync.
    return isConfigurationAcceptable(configuration, unacceptableReasons);
  }

  /**

 opendj-server-legacy/src/messages/org/opends/messages/replication.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
@@ -598,3 +599,12 @@
 in domain "%s" from this directory server DS(%d): the remote directory server DS(%d) is unknown
ERR_REPLICATION_UNEXPECTED_MESSAGE_300=New replication connection from %s started with unexpected message %s and is \
 being closed
WARN_RETRYING_BIND_CHANGELOG_301=Replication Server could not bind to the listen port : %d. \
 Error : %s. Retrying in %d ms
NOTE_BOUND_CHANGELOG_AFTER_RETRY_302=Replication Server bound to the listen port : %d \
 at attempt %d
ERR_COULD_NOT_BIND_CHANGELOG_LISTENING_303=Another socket is listening on %s
ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING_304=Nothing accepted a connection on %s : the port is held \
 either by a socket bound to another address, or by a socket which does not accept connections
ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port was released after the \
 last attempt to bind it

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

@@ -13,15 +13,31 @@
 *
 * Copyright 2006-2009 Sun Microsystems, Inc.
 * Portions Copyright 2013-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.util.StaticUtils.*;
import static org.testng.Assert.*;

import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.TestCaseUtils;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.VirtualAttributeRule;
import org.forgerock.opendj.ldap.DN;
import org.testng.annotations.Test;

@@ -47,6 +63,7 @@
      // instantiate a Replication server using the first port number.
      ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration(ports[0], null, 0, 1, 0, 0, null);
      replicationServer = new ReplicationServer(conf);
      assertTrue(replicationServer.isListening(), "the replication server should listen on port " + ports[0]);

      // Most of the configuration change are trivial to apply.
      // The interesting change is the change of the replication server port.
@@ -62,10 +79,199 @@
      // check that the sendWindow is not null to make sure that the
      // broker did connect successfully.
      assertTrue(broker.getCurrentSendWindow() != 0);
      assertTrue(replicationServer.isListening(), "the replication server should listen on port " + ports[1]);
    }
    finally
    {
      remove(replicationServer);
    }
  }

  /**
   * Tests that a replication server whose listen port cannot be bound fails fast instead
   * of silently starting without any listener, which used to surface much later, and in
   * an unrelated place, as a "connection refused", and that aborting its initialization
   * leaves the external changelog of the replication server which is already running
   * untouched: the virtual attribute rules are registered globally, by attribute name.
   */
  @Test
  public void replServerFailsWhenListenPortIsInUse() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer runningServer = null;
    try
    {
      final int[] ports = TestCaseUtils.findFreePorts(1);
      runningServer = new ReplicationServer(new ReplServerFakeConfiguration(
          ports[0], "replServerFailsWhenListenPortIsInUseRunningDb", 0, 1, 0, 0, null));
      assertTrue(runningServer.isListening());

      final List<String> rulesBefore = changelogVirtualAttributeNames();
      assertFalse(rulesBefore.isEmpty(), "the running replication server should provide the external changelog");
      final int instancesBefore = ReplicationServer.getAllInstances().size();

      // Keep the port bound for the whole lifetime of the replication server creation.
      final String dbDirName = "replServerFailsWhenListenPortIsInUseDb";
      try (ServerSocket portHolder = TestCaseUtils.bindFreePort())
      {
        final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration(
            portHolder.getLocalPort(), dbDirName, 0, 1, 0, 0, null);
        try
        {
          final ReplicationServer replicationServer = new ReplicationServer(conf);
          remove(replicationServer);
          fail("Creating a replication server on a port already in use should have failed");
        }
        catch (ConfigException expected)
        {
          // The failed replication server must not be left registered anywhere,
          assertEquals(ReplicationServer.getAllInstances().size(), instancesBefore);
          // nor must it release what it never acquired.
          assertEquals(changelogVirtualAttributeNames(), rulesBefore,
              "aborting the initialization must not deregister the virtual attribute rules"
                  + " of the running replication server");
          assertTrue(DirectoryServer.getInstance().getServerContext().getBackendConfigManager()
              .hasLocalBackend(ChangelogBackend.BACKEND_ID), "the changelog backend should still be registered");
          assertTrue(runningServer.isListening(), "the running replication server should still listen");
        }
        finally
        {
          // The aborted instance is never handed to the test, so its changelog cannot be
          // removed through ReplicationTestCase.remove().
          recursiveDelete(getFileForPath(dbDirName));
        }
      }
    }
    finally
    {
      remove(runningServer);
    }
  }

  /**
   * Tests that a listen port which is only momentarily unavailable, as it happens when a
   * socket holding it is being closed, does not prevent the replication server from
   * starting: {@code bindListenPort()} retries the bind a few times.
   * <p>
   * The port is released as soon as the replication server has actually failed to bind it,
   * so the retry is the only thing which can make it start: a test releasing the port after
   * a delay would silently stop exercising the retry as soon as the replication server took
   * longer than that delay to reach its first attempt.
   */
  @Test
  public void replServerRetriesToBindItsListenPort() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer replicationServer = null;
    final ServerSocket portHolder = TestCaseUtils.bindFreePort();
    final int bindFailuresBefore = ReplicationServer.listenPortBindFailures.get();
    try
    {
      final Thread portReleaser = new Thread(() -> {
        try
        {
          final long deadline = System.currentTimeMillis() + 30000;
          while (ReplicationServer.listenPortBindFailures.get() == bindFailuresBefore
              && System.currentTimeMillis() < deadline)
          {
            Thread.sleep(10);
          }
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
        }
        close(portHolder);
      }, "port releaser of replServerRetriesToBindItsListenPort");
      portReleaser.start();

      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          portHolder.getLocalPort(), "replServerRetriesToBindItsListenPortDb", 0, 1, 0, 0, null));
      portReleaser.join();

      assertTrue(ReplicationServer.listenPortBindFailures.get() > bindFailuresBefore,
          "the replication server should have failed its first attempt to bind the port,"
              + " otherwise this test does not exercise the retry");
      assertTrue(replicationServer.isListening(),
          "the replication server should have bound the port which was released while it was retrying");
    }
    finally
    {
      close(portHolder);
      remove(replicationServer);
    }
  }

  /**
   * Tests that a port change to a port which is not available is rejected, and that a
   * replication server which nevertheless goes through the change keeps its listen port
   * and its whole configuration: the new port is bound before the current one is released,
   * so a failure has nothing to roll back and leaves nothing half applied.
   */
  @Test
  public void replServerKeepsItsConfigurationWhenAPortChangeFails() throws Exception
  {
    TestCaseUtils.startServer();

    ReplicationServer replicationServer = null;
    try
    {
      final int[] ports = TestCaseUtils.findFreePorts(1);
      final String dbDirName = "replServerKeepsItsConfigurationWhenAPortChangeFailsDb";
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          ports[0], dbDirName, 0, 1, 0, 0, null, 1, 2000, 5000, 1));
      assertTrue(replicationServer.isListening());

      try (ServerSocket portHolder = TestCaseUtils.bindFreePort())
      {
        // The weight changes too, so that a failed change can be seen not to have applied
        // the part of the new configuration which does not depend on the listen port.
        final ReplServerFakeConfiguration newConf = new ReplServerFakeConfiguration(
            portHolder.getLocalPort(), dbDirName, 0, 1, 0, 0, null, 1, 2000, 5000, 2);

        final List<LocalizableMessage> unacceptableReasons = new ArrayList<>();
        assertFalse(replicationServer.isConfigurationChangeAcceptable(newConf, unacceptableReasons),
            "a change to a listen port which is in use should not be acceptable");
        assertFalse(unacceptableReasons.isEmpty(), "the rejected change should say why it was rejected");

        final ConfigChangeResult ccr = replicationServer.applyConfigurationChange(newConf);
        assertEquals(ccr.getResultCode(), ResultCode.OPERATIONS_ERROR);
        assertFalse(ccr.getMessages().isEmpty(), "the failed change should say why it failed");
        assertEquals(replicationServer.getReplicationPort(), ports[0],
            "the replication server should have kept its previous listen port");
        assertTrue(replicationServer.isListening(), "the replication server should still listen");
        assertEquals(replicationServer.getWeight(), 1,
            "a failed port change must not apply the rest of the new configuration");
      }

      // and it must still be usable on its original port.
      ReplicationBroker broker = openReplicationSession(
          DN.valueOf(TEST_ROOT_DN_STRING), 1, 10, ports[0], 1000);
      assertTrue(broker.getCurrentSendWindow() != 0);
    }
    finally
    {
      remove(replicationServer);
    }
  }

  /** Returns the names of the virtual attributes provided by the external changelog. */
  private List<String> changelogVirtualAttributeNames()
  {
    final Collection<String> changelogAttributes = Arrays.asList(
        "lastexternalchangelogcookie", "firstchangenumber", "lastchangenumber", "changelog");
    final List<String> names = new ArrayList<>();
    for (VirtualAttributeRule rule : DirectoryServer.getVirtualAttributes())
    {
      final String name = rule.getAttributeType().getNameOrOID().toLowerCase();
      if (changelogAttributes.contains(name))
      {
        names.add(name);
      }
    }
    Collections.sort(names);
    return names;
  }
}