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..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 @@ -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 @@ 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); } + + 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 @@ 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. 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 29ba80d77f..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; @@ -92,7 +94,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; @@ -109,8 +122,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(); @@ -124,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(); @@ -177,15 +208,25 @@ 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()); + 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 @@ 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 +498,8 @@ private void initialize() 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 @@ private void initialize() logger.trace("RS " + getMonitorInstanceName() + " creates listen thread"); } - listenThread = new ReplicationServerListenThread(this); - listenThread.start(); + startListenThread(listenSocket); if (logger.isTraceEnabled()) { @@ -476,13 +522,264 @@ private void initialize() } } 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. + *

+ * {@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. + *

+ * 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. + *

+ * 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. + *

+ * 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. *

@@ -515,6 +812,9 @@ private void enableExternalChangeLog() throws ConfigException 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 @@ 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 @@ -863,17 +1170,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; } } @@ -884,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); @@ -915,39 +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()) - { - 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 @@ private void broadcastConfigChange() public boolean isConfigurationChangeAcceptable( ReplicationServerCfg configuration, List 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 917e964de8..a6457a780b 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 @@ 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 +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 35856d53bd..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 @@ -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 @@ 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 +79,199 @@ 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", 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 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. + *

+ * 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 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 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)) + { + names.add(name); + } + } + Collections.sort(names); + return names; + } }