From 5380e8e895c410575fb9e07202992d31ed7cd8af Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 31 Jul 2026 17:00:18 +0000
Subject: [PATCH] [#792] Fail fast when the replication server cannot bind its listen port (#795)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java | 407 +++++++++++++++++++++++++++++++-----
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 10
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplicationServerListener.java | 26 +
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java | 206 ++++++++++++++++++
4 files changed, 584 insertions(+), 65 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplicationServerListener.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplicationServerListener.java
index d7d6d87..5f07c95 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplicationServerListener.java
+++ b/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;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
index b858f94..d94f169 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
+++ b/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);
}
/**
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index 917e964..a6457a7 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java
index 35856d5..ff20e85 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java
+++ b/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;
+ }
}
--
Gitblit v1.10.0