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 +++++++++++++++++++++++++++++++++++++++++++++++++--------
1 files changed, 347 insertions(+), 60 deletions(-)
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);
}
/**
--
Gitblit v1.10.0