From 5d76319b225e37231358894265b53921c315d76a Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 30 Jul 2026 16:24:40 +0300 Subject: [PATCH 1/3] [#792] Fail fast when the replication server cannot bind its listen port ReplicationServer.initialize() only logged ERR_COULD_NOT_BIND_CHANGELOG and returned normally, so a replication server whose listen port was taken was created without any listener: consumers only learned about it much later, and in an unrelated place, as a "connection refused". The same hole existed on the port change path of applyConfigurationChange(), which reported the failure as ERR_COULD_NOT_CLOSE_THE_SOCKET and left the server without a listen thread. - bind the listen port through bindListenPort(), which retries a few times with a short back-off so that a momentarily occupied port is self-healing; - propagate the failure as a ConfigException instead of swallowing it, and release the changelog DB and the external changelog when start-up is aborted; - report the failure of a port change through the ConfigChangeResult, and validate the new port in isConfigurationChangeAcceptable(), which used to accept every change unconditionally; - when a bind fails, probe the port and log whether it is owned by a live listener or by a socket which does not accept connections; - add ReplicationServer.isListening() and cover both the nominal and the port-in-use cases in ReplicationServerDynamicConfTest. Fixes #792 --- .../plugin/ReplicationServerListener.java | 6 +- .../replication/server/ReplicationServer.java | 177 ++++++++++++++++-- .../opends/messages/replication.properties | 5 + .../ReplicationServerDynamicConfTest.java | 36 ++++ 4 files changed, 211 insertions(+), 13 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 d7d6d87eac..839c0dd184 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; @@ -78,9 +79,10 @@ public ConfigChangeResult applyConfigurationAdd(ReplicationServerCfg cfg) } 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. 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 29ba80d77f..4730a68810 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 @@ -92,7 +92,18 @@ public class ReplicationServer { 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; @@ -185,7 +196,17 @@ public ReplicationServer(final ReplicationServerCfg cfg, final DSRSShutdownSync this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite); replSessionSecurity = new ReplSessionSecurity(); - initialize(); + try + { + initialize(); + } + catch (ConfigException e) + { + // This instance is never returned to the caller, so nothing will ever shut it down: + // release what initialize() managed to acquire before failing. + abortInitialization(); + throw e; + } cfg.addChangeListener(this); localPorts.add(getReplicationPort()); @@ -441,8 +462,14 @@ private boolean connect(HostPort remoteServerAddress, DN baseDN) 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 +478,7 @@ private void initialize() this.changelogDB.initializeDB(); setServerURL(); - listenSocket = new ServerSocket(); - listenSocket.bind(new InetSocketAddress(getReplicationPort())); + listenSocket = bindListenPort(); // creates working threads: we must first connect, then start to listen. if (logger.isTraceEnabled()) @@ -477,12 +503,120 @@ private void initialize() } catch (UnknownHostException e) { logger.error(ERR_UNKNOWN_HOSTNAME); + 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. + final LocalizableMessage message = + ERR_COULD_NOT_BIND_CHANGELOG.get(getReplicationPort(), describeBindFailure(getReplicationPort(), e)); + logger.error(message); + throw new ConfigException(message, e); } } + /** + * Binds the listen port, retrying a few times when the port is momentarily unavailable. + * + * @return the bound listen socket + * @throws IOException + * if the port could not be bound within {@link #LISTEN_BIND_ATTEMPTS} attempts + */ + private ServerSocket bindListenPort() throws IOException + { + final int port = getReplicationPort(); + 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); + if (attempt >= LISTEN_BIND_ATTEMPTS) + { + throw e; + } + logger.warn(WARN_RETRYING_BIND_CHANGELOG, port, describeBindFailure(port, e), LISTEN_BIND_RETRY_DELAY_MS); + try + { + Thread.sleep(LISTEN_BIND_RETRY_DELAY_MS); + } + catch (InterruptedException e2) + { + Thread.currentThread().interrupt(); + throw e; + } + } + } + } + + /** + * Returns a description of a failed bind of the listen port, i.e. the error itself plus a + * best effort diagnostic of what holds the port. + *

+ * The port is probed over the loopback interface, which covers the realistic cases of a + * wildcard or loopback listener. A successful connect means a live listener owns the + * port, while a refused connect means the port is held by a socket which does not accept + * connections, e.g. a client socket which was given that port as its local port. + * + * @param port + * the port which could not be bound + * @param cause + * the error returned by the failed bind + * @return a human readable description of the failure + */ + private static String describeBindFailure(int port, IOException cause) + { + final String diagnostic; + try (Socket probe = new Socket()) + { + probe.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), LISTEN_PORT_PROBE_TIMEOUT_MS); + diagnostic = isSelfConnection(probe) + ? "the port is held by a TCP self-connect, nothing is listening on it" + : "another socket is listening on the port"; + } + catch (IOException e) + { + return cause.getMessage() + " (the port is held by a socket which does not accept connections)"; + } + return cause.getMessage() + " (" + diagnostic + ")"; + } + + /** Releases what {@link #initialize()} acquired before it failed. */ + private void abortInitialization() + { + close(listenSocket); + shutdownExternalChangelog(); + 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. *

@@ -930,21 +1064,36 @@ public ConfigChangeResult applyConfigurationChange( stopListen = false; setServerURL(); - listenSocket = new ServerSocket(); - listenSocket.bind(new InetSocketAddress(getReplicationPort())); + listenSocket = bindListenPort(); listenThread = new ReplicationServerListenThread(this); listenThread.start(); } + catch (UnknownHostException e) + { + logger.traceException(e); + logger.error(ERR_UNKNOWN_HOSTNAME); + ccr.setResultCode(ResultCode.OPERATIONS_ERROR); + ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get()); + } catch (IOException e) { + // The old listen socket is already closed, so this replication server is now left + // without any listener: report the failure instead of silently going deaf. logger.traceException(e); - logger.error(ERR_COULD_NOT_CLOSE_THE_SOCKET, e); + final LocalizableMessage message = + ERR_COULD_NOT_BIND_CHANGELOG.get(getReplicationPort(), describeBindFailure(getReplicationPort(), e)); + logger.error(message); + ccr.setResultCode(ResultCode.OPERATIONS_ERROR); + ccr.addMessage(message); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); logger.traceException(e); logger.error(ERR_COULD_NOT_STOP_LISTEN_THREAD, e); + ccr.setResultCode(ResultCode.OPERATIONS_ERROR); + ccr.addMessage(ERR_COULD_NOT_STOP_LISTEN_THREAD.get(e)); } } @@ -1031,7 +1180,13 @@ private void broadcastConfigChange() public boolean isConfigurationChangeAcceptable( ReplicationServerCfg configuration, List unacceptableReasons) { - return true; + if (configuration.getReplicationPort() == getReplicationPort()) + { + return true; + } + // applyConfigurationChange() closes the current listen socket before binding the new + // port and cannot roll back to the old one, so the new port must be available. + 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 917e964de8..ad96441257 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,7 @@ ERR_FULL_UPDATE_MISSING_REMOTE_299=Cannot start total update \ 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 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 35856d53bd..59efcda294 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,12 +13,16 @@ * * 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.testng.Assert.*; +import java.net.ServerSocket; + +import org.forgerock.opendj.config.server.ConfigException; import org.opends.server.TestCaseUtils; import org.opends.server.replication.ReplicationTestCase; import org.opends.server.replication.service.ReplicationBroker; @@ -47,6 +51,7 @@ public void replServerApplyChangeTest() throws Exception // 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 +67,41 @@ public void replServerApplyChangeTest() throws Exception // 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". + */ + @Test + public void replServerFailsWhenListenPortIsInUse() throws Exception + { + TestCaseUtils.startServer(); + + final int instancesBefore = ReplicationServer.getAllInstances().size(); + // Keep the port bound for the whole lifetime of the replication server creation. + try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) + { + final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration( + portHolder.getLocalPort(), "replServerFailsWhenListenPortIsInUseDb", 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); + } + } + } } From c47f8673451d37897183d155380a89d4d3588bcd Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 31 Jul 2026 10:46:35 +0300 Subject: [PATCH 2/3] [#792] Address review feedback on the fail-fast replication bind - set SO_REUSEADDR on the diagnostic probe, as every other socket which can land on a replication port does: a probe which self-connects would otherwise hold the port in TIME_WAIT once closed and defeat the retry it diagnoses; - probe the port once, when giving up, instead of on every attempt, and report only what was observed on the loopback interface: a listener bound to a non loopback address used to be reported as a socket which does not accept connections, i.e. the opposite of the truth; - give the diagnostics their own localizable messages instead of injecting English text into the %s of a localized one; - guard the deregistration of the external changelog virtual attribute rules: they are registered globally, by attribute name, so an aborted instance used to strip them from a running one; - create the replication server before registering the configuration listeners, which were leaked in the server wide configuration repository when the creation failed, and log why dsconfig create-replication-server failed; - roll back to the previous listen port when a port change fails, keep localPorts in sync, and only clear stopListen once the port is bound, so an interrupted change can no longer leave a listen thread which exits at once; - close the socket of isConfigurationAcceptable() on the failure path too; - stop the threads and release the changelog when initialization is aborted; - make stopListen volatile, log the bind failure once instead of twice; - cover the retry, the aborted initialization and the failed port change with tests. --- .../plugin/ReplicationServerListener.java | 13 +- .../replication/server/ReplicationServer.java | 182 ++++++++++++++---- .../opends/messages/replication.properties | 7 + .../ReplicationServerDynamicConfTest.java | 172 +++++++++++++++-- 4 files changed, 316 insertions(+), 58 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 839c0dd184..a0bb9f1571 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 @@ -20,6 +20,7 @@ import java.util.List; import org.forgerock.i18n.LocalizableMessage; +import org.forgerock.i18n.slf4j.LocalizedLogger; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.ResultCode; import org.forgerock.opendj.config.server.ConfigurationAddListener; @@ -39,6 +40,8 @@ public class ReplicationServerListener implements ConfigurationAddListener, ConfigurationDeleteListener { + private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); + private final DSRSShutdownSync dsrsShutdownSync; private ReplicationServer replicationServer; @@ -57,15 +60,18 @@ public ReplicationServerListener( 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); } + + configuration.addReplicationServerAddListener(this); + configuration.addReplicationServerDeleteListener(this); } /** {@inheritDoc} */ @@ -81,6 +87,7 @@ public ConfigChangeResult applyConfigurationAdd(ReplicationServerCfg cfg) { // The configEntry has already been validated in configAddisAcceptable, but the // listen port may have been taken since then: report why the creation failed. + logger.error(e.getMessageObject()); ccr.setResultCode(ResultCode.CONSTRAINT_VIOLATION); ccr.addMessage(e.getMessageObject()); } 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 4730a68810..4519117be6 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 @@ -120,8 +120,16 @@ public class ReplicationServer /** 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(); @@ -502,16 +510,15 @@ private void initialize() throws ConfigException } } 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) { // 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. - final LocalizableMessage message = - ERR_COULD_NOT_BIND_CHANGELOG.get(getReplicationPort(), describeBindFailure(getReplicationPort(), e)); - logger.error(message); - throw new ConfigException(message, e); + logger.traceException(e); + throw new ConfigException(bindFailureMessage(getReplicationPort(), e), e); } } @@ -544,7 +551,9 @@ private ServerSocket bindListenPort() throws IOException { throw e; } - logger.warn(WARN_RETRYING_BIND_CHANGELOG, port, describeBindFailure(port, e), LISTEN_BIND_RETRY_DELAY_MS); + // 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); @@ -559,41 +568,129 @@ private ServerSocket bindListenPort() throws IOException } /** - * Returns a description of a failed bind of the listen port, i.e. the error itself plus a + * 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; + } + } + + /** + * Binds the listen port of the current configuration and starts a listen thread on it. + *

+ * {@code stopListen} is only cleared once the port is bound, so a failure leaves this + * replication server consistently stopped rather than with a listen thread which would + * spin on a closed socket. + * + * @throws IOException + * if the listen port could not be bound + */ + private void startListenThread() throws IOException + { + setServerURL(); + listenSocket = bindListenPort(); + + stopListen = false; + listenThread = new ReplicationServerListenThread(this); + listenThread.start(); + } + + /** + * Goes back to the listen port of the previous configuration after a port change failed, + * so that this replication server does not stay deaf with a configuration which + * advertises a port it does not listen to. + * + * @param previousConfig + * the configuration which was in use before the failed change + * @param ccr + * the result of the configuration change, to which a failed rollback is added + */ + private void rollbackListenPort(ReplicationServerCfg previousConfig, ConfigChangeResult ccr) + { + this.config = previousConfig; + try + { + startListenThread(); + } + catch (UnknownHostException e) + { + logger.traceException(e); + logger.error(ERR_UNKNOWN_HOSTNAME); + ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get()); + } + catch (IOException e) + { + // Nothing else can be done: the port was free a moment ago, it is not anymore. + logger.traceException(e); + final LocalizableMessage message = bindFailureMessage(getReplicationPort(), e); + logger.error(message); + ccr.addMessage(message); + } + } + + /** + * 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. *

- * The port is probed over the loopback interface, which covers the realistic cases of a - * wildcard or loopback listener. A successful connect means a live listener owns the - * port, while a refused connect means the port is held by a socket which does not accept - * connections, e.g. a client socket which was given that port as its local port. + * 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 * @param cause * the error returned by the failed bind - * @return a human readable description of the failure + * @return the message describing the failure */ - private static String describeBindFailure(int port, IOException cause) + private static LocalizableMessage bindFailureMessage(int port, IOException cause) { - final String diagnostic; + final LocalizableMessage error = getExceptionMessage(cause); + 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); - diagnostic = isSelfConnection(probe) - ? "the port is held by a TCP self-connect, nothing is listening on it" - : "another socket is listening on the port"; + 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(port, error, probedAddress); + } + return ERR_COULD_NOT_BIND_CHANGELOG_LISTENING.get(port, error, probedAddress); } catch (IOException e) { - return cause.getMessage() + " (the port is held by a socket which does not accept connections)"; + logger.traceException(e); + return ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING.get(port, error, probedAddress); } - return cause.getMessage() + " (" + diagnostic + ")"; } /** Releases what {@link #initialize()} acquired before it failed. */ private void abortInitialization() { + shutdown.set(true); + if (connectThread != null) + { + connectThread.interrupt(); + } close(listenSocket); + if (listenThread != null) + { + listenThread.interrupt(); + } shutdownExternalChangelog(); try { @@ -650,6 +747,7 @@ private void enableExternalChangeLog() throws ConfigException } registerVirtualAttributeRules(); + externalChangelogRegistered = true; } catch (Exception e) { @@ -669,7 +767,14 @@ private void shutdownExternalChangelog() 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 getVirtualAttributesRules() throws DirectoryException @@ -997,17 +1102,14 @@ public static boolean isConfigurationAcceptable( { 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; } } @@ -1053,21 +1155,13 @@ public ConfigChangeResult applyConfigurationChange( // and restart it. if (getReplicationPort() != oldConfig.getReplicationPort()) { - stopListen = true; try { - close(listenSocket); - if (listenThread != null) - { - listenThread.join(); - } - stopListen = false; - - setServerURL(); - listenSocket = bindListenPort(); + stopListenThread(); + startListenThread(); - listenThread = new ReplicationServerListenThread(this); - listenThread.start(); + localPorts.remove(oldConfig.getReplicationPort()); + localPorts.add(getReplicationPort()); } catch (UnknownHostException e) { @@ -1075,25 +1169,29 @@ public ConfigChangeResult applyConfigurationChange( logger.error(ERR_UNKNOWN_HOSTNAME); ccr.setResultCode(ResultCode.OPERATIONS_ERROR); ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get()); + rollbackListenPort(oldConfig, ccr); } catch (IOException e) { - // The old listen socket is already closed, so this replication server is now left - // without any listener: report the failure instead of silently going deaf. + // The old listen socket is already closed, so this replication server would be left + // without any listener: report the failure and go back to the previous port. logger.traceException(e); - final LocalizableMessage message = - ERR_COULD_NOT_BIND_CHANGELOG.get(getReplicationPort(), describeBindFailure(getReplicationPort(), e)); + final LocalizableMessage message = bindFailureMessage(getReplicationPort(), e); logger.error(message); ccr.setResultCode(ResultCode.OPERATIONS_ERROR); ccr.addMessage(message); + rollbackListenPort(oldConfig, ccr); } catch (InterruptedException e) { + // The previous listen thread may still be running, so do not start another one: + // stopListen is still set, which makes that thread stop as soon as it wakes up. Thread.currentThread().interrupt(); logger.traceException(e); logger.error(ERR_COULD_NOT_STOP_LISTEN_THREAD, e); ccr.setResultCode(ResultCode.OPERATIONS_ERROR); ccr.addMessage(ERR_COULD_NOT_STOP_LISTEN_THREAD.get(e)); + this.config = oldConfig; } } 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 ad96441257..bb142272be 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -603,3 +603,10 @@ WARN_RETRYING_BIND_CHANGELOG_301=Replication Server could not bind to the listen 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=Replication Server could not bind to the listen port : %d. \ + Error : %s. Another socket is listening on %s +ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING_304=Replication Server could not bind to the listen port : %d. \ + Error : %s. 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=Replication Server could not bind to the listen port : %d. \ + Error : %s. 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 59efcda294..9f5bd0ccde 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 @@ -18,14 +18,26 @@ 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; @@ -78,30 +90,164 @@ public void replServerApplyChangeTest() throws Exception /** * 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". + * 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(); - final int instancesBefore = ReplicationServer.getAllInstances().size(); - // Keep the port bound for the whole lifetime of the replication server creation. - try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) + ReplicationServer runningServer = null; + try { - final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration( - portHolder.getLocalPort(), "replServerFailsWhenListenPortIsInUseDb", 0, 1, 0, 0, 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 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. + try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) + { + final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration( + portHolder.getLocalPort(), "replServerFailsWhenListenPortIsInUseDb", 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 + { + 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. + */ + @Test + public void replServerRetriesToBindItsListenPort() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + final ServerSocket portHolder = TestCaseUtils.bindFreePort(); + try + { + // Release the port while the replication server is retrying to bind it. The retry + // gives 800 ms in total, which leaves a wide margin for this thread to be scheduled. + final Thread portReleaser = new Thread(() -> { + try + { + Thread.sleep(300); + } + 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.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 goes back to its + * previous listen port instead of staying deaf with a configuration which advertises a + * port it does not listen to. + */ + @Test + public void replServerRollsBackAFailedPortChange() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + try + { + final int[] ports = TestCaseUtils.findFreePorts(1); + replicationServer = new ReplicationServer(new ReplServerFakeConfiguration( + ports[0], "replServerRollsBackAFailedPortChangeDb", 0, 1, 0, 0, null)); + assertTrue(replicationServer.isListening()); + + try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) { - final ReplicationServer replicationServer = new ReplicationServer(conf); - remove(replicationServer); - fail("Creating a replication server on a port already in use should have failed"); + final ReplServerFakeConfiguration newConf = new ReplServerFakeConfiguration( + portHolder.getLocalPort(), "replServerRollsBackAFailedPortChangeDb", 0, 1, 0, 0, null); + + final List 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 gone back to its previous listen port"); + assertTrue(replicationServer.isListening(), "the replication server should still listen"); } - catch (ConfigException expected) + + // 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 changelogVirtualAttributeNames() + { + final Collection changelogAttributes = Arrays.asList( + "lastexternalchangelogcookie", "firstchangenumber", "lastchangenumber", "changelog"); + final List names = new ArrayList<>(); + for (VirtualAttributeRule rule : DirectoryServer.getVirtualAttributes()) + { + final String name = rule.getAttributeType().getNameOrOID().toLowerCase(); + if (changelogAttributes.contains(name)) { - // The failed replication server must not be left registered anywhere. - assertEquals(ReplicationServer.getAllInstances().size(), instancesBefore); + names.add(name); } } + Collections.sort(names); + return names; } } From 07cab8511b4421109ea22406c5037c8a4e6d79f0 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 31 Jul 2026 13:20:00 +0300 Subject: [PATCH 3/3] [#792] Bind the new listen port before releasing the current one A failed port change used to leave a half applied configuration: the purge delay, compute-change-number, the crypto parameters and the disconnection of removed peers were applied from the new configuration, while rollbackListenPort() restored only this.config, which made every block after the port block compare an object with itself and no-op. applyConfigurationChange() now switches the listen port first, before applying anything else, and switchListenPort() binds the new port while the current listen socket is still open and serving. A change which cannot be applied therefore leaves the replication server exactly as it was: nothing to roll back, no window without a listener, and no window during which the server URL advertises a port nothing listens to. An interrupted stop of the listen thread closes the socket bound for the new port and keeps stopListen set, so no second listen thread can accept on it. Also: * ReplicationServerListener shuts the replication server down when registering the configuration listeners fails, instead of leaving a started server nothing can stop; * the explicit logger.error() calls on the ConfigChangeResult paths are removed, as ConfigurationHandler.handleConfigChangeResult() already logs the result messages; * the bind failure message is built again on the translated ERR_COULD_NOT_BIND_CHANGELOG, with the loopback diagnostic appended as its own message key; * enableExternalChangeLog() runs inside the try which aborts the initialization, and externalChangelogRegistered is set before the virtual attribute rules are registered, so a partial registration is released too; * replServerRetriesToBindItsListenPort releases the port when the bind has actually failed, using the new listenPortBindFailures counter, and asserts the retry was exercised. --- .../plugin/ReplicationServerListener.java | 21 +- .../replication/server/ReplicationServer.java | 242 ++++++++++-------- .../opends/messages/replication.properties | 12 +- .../ReplicationServerDynamicConfTest.java | 46 +++- 4 files changed, 192 insertions(+), 129 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 a0bb9f1571..5f07c95b29 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 @@ -20,7 +20,6 @@ import java.util.List; import org.forgerock.i18n.LocalizableMessage; -import org.forgerock.i18n.slf4j.LocalizedLogger; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.ResultCode; import org.forgerock.opendj.config.server.ConfigurationAddListener; @@ -40,8 +39,6 @@ public class ReplicationServerListener implements ConfigurationAddListener, ConfigurationDeleteListener { - private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); - private final DSRSShutdownSync dsrsShutdownSync; private ReplicationServer replicationServer; @@ -70,8 +67,18 @@ public ReplicationServerListener( replicationServer = new ReplicationServer(cfg, dsrsShutdownSync); } - configuration.addReplicationServerAddListener(this); - configuration.addReplicationServerDeleteListener(this); + 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} */ @@ -86,8 +93,8 @@ public ConfigChangeResult applyConfigurationAdd(ReplicationServerCfg cfg) catch (ConfigException e) { // The configEntry has already been validated in configAddisAcceptable, but the - // listen port may have been taken since then: report why the creation failed. - logger.error(e.getMessageObject()); + // 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()); } 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 4519117be6..37ebbf4695 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; @@ -143,6 +145,16 @@ public class ReplicationServer */ private static final Set localPorts = new CopyOnWriteArraySet<>(); + /** + * Number of attempts to bind a listen port which failed in this VM. + *

+ * 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(); @@ -196,22 +208,22 @@ public ReplicationServer(final ReplicationServerCfg cfg, final DSRSShutdownSync this.dsrsShutdownSync = dsrsShutdownSync; this.domainPredicate = predicate; - enableExternalChangeLog(); - ServerContext serverContext = DirectoryServer.getInstance().getServerContext(); - cryptoSuite = serverContext.getCryptoManager(). - newCryptoSuite(cfg.getCipherTransformation(), cfg.getCipherKeyLength(), cfg.isConfidentialityEnabled()); - - this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite); - - replSessionSecurity = new ReplSessionSecurity(); 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); + + replSessionSecurity = new ReplSessionSecurity(); initialize(); } catch (ConfigException e) { // This instance is never returned to the caller, so nothing will ever shut it down: - // release what initialize() managed to acquire before failing. + // release what the initialization managed to acquire before failing. abortInitialization(); throw e; } @@ -486,7 +498,8 @@ private void initialize() throws ConfigException this.changelogDB.initializeDB(); setServerURL(); - listenSocket = bindListenPort(); + // 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()) @@ -501,8 +514,7 @@ private void initialize() throws ConfigException logger.trace("RS " + getMonitorInstanceName() + " creates listen thread"); } - listenThread = new ReplicationServerListenThread(this); - listenThread.start(); + startListenThread(listenSocket); if (logger.isTraceEnabled()) { @@ -523,15 +535,16 @@ private void initialize() throws ConfigException } /** - * Binds the listen port, retrying a few times when the port is momentarily unavailable. + * 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() throws IOException + private ServerSocket bindListenPort(int port) throws IOException { - final int port = getReplicationPort(); for (int attempt = 1; ; attempt++) { final ServerSocket socket = new ServerSocket(); @@ -547,6 +560,7 @@ private ServerSocket bindListenPort() throws IOException catch (IOException e) { close(socket); + listenPortBindFailures.incrementAndGet(); if (attempt >= LISTEN_BIND_ATTEMPTS) { throw e; @@ -585,61 +599,108 @@ private void stopListenThread() throws InterruptedException } /** - * Binds the listen port of the current configuration and starts a listen thread on it. + * Starts a listen thread on the provided listen socket. *

- * {@code stopListen} is only cleared once the port is bound, so a failure leaves this - * replication server consistently stopped rather than with a listen thread which would - * spin on a closed socket. + * {@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. * - * @throws IOException - * if the listen port could not be bound + * @param boundListenSocket + * the bound socket the listen thread will accept connections on */ - private void startListenThread() throws IOException + private void startListenThread(ServerSocket boundListenSocket) { - setServerURL(); - listenSocket = bindListenPort(); - + listenSocket = boundListenSocket; stopListen = false; listenThread = new ReplicationServerListenThread(this); listenThread.start(); } /** - * Goes back to the listen port of the previous configuration after a port change failed, - * so that this replication server does not stay deaf with a configuration which - * advertises a port it does not listen to. + * Switches the listen port to the one of the provided configuration. + *

+ * 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 previousConfig - * the configuration which was in use before the failed change + * @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 failed rollback is added + * 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 void rollbackListenPort(ReplicationServerCfg previousConfig, ConfigChangeResult ccr) + private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeResult ccr) { - this.config = previousConfig; + final ReplicationServerCfg previousConfig = this.config; + final String previousServerURL = serverURL; + final int newPort = newConfig.getReplicationPort(); + ServerSocket newListenSocket = null; try { - startListenThread(); + // 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); - logger.error(ERR_UNKNOWN_HOSTNAME); + ccr.setResultCode(ResultCode.OPERATIONS_ERROR); ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get()); } catch (IOException e) { - // Nothing else can be done: the port was free a moment ago, it is not anymore. + // 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); - final LocalizableMessage message = bindFailureMessage(getReplicationPort(), e); - logger.error(message); - ccr.addMessage(message); + 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. *

* 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 @@ -649,13 +710,10 @@ private void rollbackListenPort(ReplicationServerCfg previousConfig, ConfigChang * * @param port * the port which could not be bound - * @param cause - * the error returned by the failed bind - * @return the message describing the failure + * @return the message describing what was observed on the loopback interface */ - private static LocalizableMessage bindFailureMessage(int port, IOException cause) + private static LocalizableMessage describeListenPortHolder(int port) { - final LocalizableMessage error = getExceptionMessage(cause); final String probedAddress = InetAddress.getLoopbackAddress().getHostAddress() + ":" + port; try (Socket probe = new Socket()) { @@ -667,18 +725,23 @@ private static LocalizableMessage bindFailureMessage(int port, IOException cause { // 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(port, error, probedAddress); + return ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE.get(probedAddress); } - return ERR_COULD_NOT_BIND_CHANGELOG_LISTENING.get(port, error, 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(port, error, probedAddress); + return ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING.get(probedAddress); } } - /** Releases what {@link #initialize()} acquired before it failed. */ + /** + * Releases what the initialization acquired before it failed. + *

+ * It runs on a partially constructed instance, so every field it uses may still be + * unassigned. + */ private void abortInitialization() { shutdown.set(true); @@ -692,13 +755,16 @@ private void abortInitialization() listenThread.interrupt(); } shutdownExternalChangelog(); - try - { - this.changelogDB.shutdownDB(); - } - catch (ChangelogException ignored) + if (this.changelogDB != null) { - logger.traceException(ignored); + try + { + this.changelogDB.shutdownDB(); + } + catch (ChangelogException ignored) + { + logger.traceException(ignored); + } } } @@ -746,8 +812,10 @@ private void enableExternalChangeLog() throws ConfigException getExceptionMessage(e))); } - registerVirtualAttributeRules(); + // 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) { @@ -1120,11 +1188,21 @@ public ConfigChangeResult applyConfigurationChange( { final ConfigChangeResult ccr = new ConfigChangeResult(); - // Some of those properties change don't need specific code. - // They will be applied for next connections. Some others have immediate effect final Set 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 this.config = configuration; disconnectRemovedReplicationServers(oldRSAddresses); @@ -1151,50 +1229,6 @@ public ConfigChangeResult applyConfigurationChange( 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()) - { - try - { - stopListenThread(); - startListenThread(); - - localPorts.remove(oldConfig.getReplicationPort()); - localPorts.add(getReplicationPort()); - } - catch (UnknownHostException e) - { - logger.traceException(e); - logger.error(ERR_UNKNOWN_HOSTNAME); - ccr.setResultCode(ResultCode.OPERATIONS_ERROR); - ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get()); - rollbackListenPort(oldConfig, ccr); - } - catch (IOException e) - { - // The old listen socket is already closed, so this replication server would be left - // without any listener: report the failure and go back to the previous port. - logger.traceException(e); - final LocalizableMessage message = bindFailureMessage(getReplicationPort(), e); - logger.error(message); - ccr.setResultCode(ResultCode.OPERATIONS_ERROR); - ccr.addMessage(message); - rollbackListenPort(oldConfig, ccr); - } - catch (InterruptedException e) - { - // The previous listen thread may still be running, so do not start another one: - // stopListen is still set, which makes that thread stop as soon as it wakes up. - Thread.currentThread().interrupt(); - logger.traceException(e); - logger.error(ERR_COULD_NOT_STOP_LISTEN_THREAD, e); - ccr.setResultCode(ResultCode.OPERATIONS_ERROR); - ccr.addMessage(ERR_COULD_NOT_STOP_LISTEN_THREAD.get(e)); - this.config = oldConfig; - } - } - // Update period value for monitoring publishers if (oldConfig.getMonitoringPeriod() != config.getMonitoringPeriod()) { @@ -1282,8 +1316,8 @@ public boolean isConfigurationChangeAcceptable( { return true; } - // applyConfigurationChange() closes the current listen socket before binding the new - // port and cannot roll back to the old one, so the new port must be available. + // 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 bb142272be..a6457a780b 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -603,10 +603,8 @@ WARN_RETRYING_BIND_CHANGELOG_301=Replication Server could not bind to the listen 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=Replication Server could not bind to the listen port : %d. \ - Error : %s. Another socket is listening on %s -ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING_304=Replication Server could not bind to the listen port : %d. \ - Error : %s. 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=Replication Server could not bind to the listen port : %d. \ - Error : %s. Nothing holds %s anymore : the port was released after the last attempt to bind it +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 9f5bd0ccde..ff20e85ec4 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 @@ -112,10 +112,11 @@ public void replServerFailsWhenListenPortIsInUse() throws Exception 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(), "replServerFailsWhenListenPortIsInUseDb", 0, 1, 0, 0, null); + portHolder.getLocalPort(), dbDirName, 0, 1, 0, 0, null); try { final ReplicationServer replicationServer = new ReplicationServer(conf); @@ -134,6 +135,12 @@ public void replServerFailsWhenListenPortIsInUse() throws Exception .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 @@ -146,6 +153,11 @@ public void replServerFailsWhenListenPortIsInUse() throws Exception * 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. + *

+ * 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 @@ -154,14 +166,18 @@ public void replServerRetriesToBindItsListenPort() throws Exception ReplicationServer replicationServer = null; final ServerSocket portHolder = TestCaseUtils.bindFreePort(); + final int bindFailuresBefore = ReplicationServer.listenPortBindFailures.get(); try { - // Release the port while the replication server is retrying to bind it. The retry - // gives 800 ms in total, which leaves a wide margin for this thread to be scheduled. final Thread portReleaser = new Thread(() -> { try { - Thread.sleep(300); + final long deadline = System.currentTimeMillis() + 30000; + while (ReplicationServer.listenPortBindFailures.get() == bindFailuresBefore + && System.currentTimeMillis() < deadline) + { + Thread.sleep(10); + } } catch (InterruptedException e) { @@ -175,6 +191,9 @@ public void replServerRetriesToBindItsListenPort() throws Exception 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"); } @@ -187,12 +206,12 @@ public void replServerRetriesToBindItsListenPort() throws Exception /** * 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 goes back to its - * previous listen port instead of staying deaf with a configuration which advertises a - * port it does not listen to. + * 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 replServerRollsBackAFailedPortChange() throws Exception + public void replServerKeepsItsConfigurationWhenAPortChangeFails() throws Exception { TestCaseUtils.startServer(); @@ -200,14 +219,17 @@ public void replServerRollsBackAFailedPortChange() throws Exception try { final int[] ports = TestCaseUtils.findFreePorts(1); + final String dbDirName = "replServerKeepsItsConfigurationWhenAPortChangeFailsDb"; replicationServer = new ReplicationServer(new ReplServerFakeConfiguration( - ports[0], "replServerRollsBackAFailedPortChangeDb", 0, 1, 0, 0, null)); + 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(), "replServerRollsBackAFailedPortChangeDb", 0, 1, 0, 0, null); + portHolder.getLocalPort(), dbDirName, 0, 1, 0, 0, null, 1, 2000, 5000, 2); final List unacceptableReasons = new ArrayList<>(); assertFalse(replicationServer.isConfigurationChangeAcceptable(newConf, unacceptableReasons), @@ -218,8 +240,10 @@ public void replServerRollsBackAFailedPortChange() throws Exception 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 gone back to its previous listen port"); + "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.