From 1b1c01a9bb1b1d410ae0c2635fc21e9a456b374c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 31 Jul 2026 14:59:47 +0300 Subject: [PATCH 1/5] [#802] Fail fast when the replication server cannot read its changelog FileChangelogDB.initializeDB() caught ChangelogException and only logged ERR_COULD_NOT_READ_DB, whose text already says the replication server failed to start. It did start: ReplicationServer.initialize() went on to bind the listen port and start its threads over a changelog it never opened, so the failure surfaced much later and somewhere else - as a failure on the first update to be persisted when the replication environment does not exist at all, or as a domain adopting the generation id of the first replica to connect over a changelog which holds another generation. initializeDB() now declares ChangelogException, FileChangelogDB wraps the cause in the same localized ERR_COULD_NOT_READ_DB message and rethrows it, and initialize() reports it as a ConfigException, exactly like a listen port which cannot be bound. The constructor already releases a half initialized instance through abortInitialization(), which shuts the changelog DB down. ReplicationServerDynamicConfTest.replServerFailsWhenChangelogCannotBeRead covers it: a corrupted domains.state makes the creation fail with a ConfigException naming the changelog directory, leaves no instance registered and no listen port bound. --- .../replication/server/ReplicationServer.java | 11 +++- .../server/changelog/api/ChangelogDB.java | 8 ++- .../changelog/file/FileChangelogDB.java | 11 +++- .../ReplicationServerDynamicConfTest.java | 60 +++++++++++++++++++ 4 files changed, 85 insertions(+), 5 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index d94f169e48..14e8fb11d4 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 @@ -486,8 +486,8 @@ private boolean connect(HostPort remoteServerAddress, DN baseDN) * Initialization function for the replicationServer. * * @throws ConfigException - * when the replication server cannot be started, in particular when its listen - * port cannot be bound. + * when the replication server cannot be started, in particular when its changelog + * cannot be read or when its listen port cannot be bound. */ private void initialize() throws ConfigException { @@ -520,6 +520,13 @@ private void initialize() throws ConfigException { logger.trace("RS " + getMonitorInstanceName() + " successfully initialized"); } + } catch (ChangelogException e) + { + // A replication server which cannot read its changelog is as dead as one which cannot + // bind its listen port: it would otherwise accept connections and fail on the first + // change it has to persist. The message already names the changelog directory. + logger.traceException(e); + throw new ConfigException(e.getMessageObject(), e); } catch (UnknownHostException e) { // Not logged here: the caller reports the ConfigException, logging it once. diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java index 9b9e3efe39..a75f7dad18 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java @@ -12,6 +12,7 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2013 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server.changelog.api; @@ -29,8 +30,13 @@ public interface ChangelogDB * Initializes the replication database by reading its previous state and * building the relevant ReplicaDBs according to the previous state. This * method must be called once before using the ChangelogDB. + * + * @throws ChangelogException + * If the previous state could not be read. The database is then + * unusable, possibly half open, and the caller must release it by + * calling {@link #shutdownDB()}. */ - void initializeDB(); + void initializeDB() throws ChangelogException; /** * Sets the purge delay for the replication database. Can be called while the diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 2164ae965f..ce6a9fdcbb 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -279,7 +279,7 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM } @Override - public void initializeDB() + public void initializeDB() throws ChangelogException { try { @@ -294,8 +294,15 @@ public void initializeDB() } catch (ChangelogException e) { + // A changelog which could not be read leaves this DB unusable: the replication + // environment may not exist at all, and the state which was restored before the failure + // only covers part of the domains. Both surface much later and somewhere else: as a + // failure on the first update to be persisted, or as a domain adopting the generation id + // of the first replica to connect, over a changelog which holds another generation. + // Not logged here: the caller reports the failure, logging it once. logger.traceException(e); - logger.error(ERR_COULD_NOT_READ_DB, this.dbDirectory.getAbsolutePath(), e.getLocalizedMessage()); + throw new ChangelogException( + ERR_COULD_NOT_READ_DB.get(this.dbDirectory.getAbsolutePath(), e.getLocalizedMessage()), e); } } 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 ff20e85ec4..52e3b9c2fd 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 @@ -21,7 +21,11 @@ import static org.opends.server.util.StaticUtils.*; import static org.testng.Assert.*; +import java.io.File; +import java.net.InetSocketAddress; import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -36,6 +40,7 @@ import org.opends.server.backends.ChangelogBackend; import org.opends.server.core.DirectoryServer; import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.server.changelog.api.ChangelogException; import org.opends.server.replication.service.ReplicationBroker; import org.opends.server.types.VirtualAttributeRule; import org.forgerock.opendj.ldap.DN; @@ -257,6 +262,61 @@ public void replServerKeepsItsConfigurationWhenAPortChangeFails() throws Excepti } } + /** + * Tests that a replication server whose changelog cannot be read fails fast instead of + * starting over a changelog it never opened: it used to log ERR_COULD_NOT_READ_DB, whose + * text already says the replication server failed to start, then bind its listen port and + * accept connections anyway, so the failure surfaced much later and somewhere else. + */ + @Test + public void replServerFailsWhenChangelogCannotBeRead() throws Exception + { + TestCaseUtils.startServer(); + + final String dbDirName = "replServerFailsWhenChangelogCannotBeReadDb"; + final File dbDirectory = getFileForPath(dbDirName); + try + { + // A domains.state whose second field is not a DN: what a corrupted changelog state file + // looks like to ReplicationEnvironment, which then cannot be created at all. + assertTrue(dbDirectory.isDirectory() || dbDirectory.mkdirs(), "could not create " + dbDirectory); + Files.write(new File(dbDirectory, "domains.state").toPath(), + Collections.singletonList("1:this is not a DN"), StandardCharsets.UTF_8); + + final int[] ports = TestCaseUtils.findFreePorts(1); + final int instancesBefore = ReplicationServer.getAllInstances().size(); + try + { + final ReplicationServer replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, 1, 0, 0, null)); + remove(replicationServer); + fail("Creating a replication server over an unreadable changelog should have failed"); + } + catch (ConfigException expected) + { + assertTrue(expected.getCause() instanceof ChangelogException, + "the failure should be the one of the changelog, but was: " + expected.getCause()); + assertTrue(expected.getMessage().contains(dbDirectory.getAbsolutePath()), + "the failure should name the changelog directory, but was: " + expected.getMessage()); + assertEquals(ReplicationServer.getAllInstances().size(), instancesBefore, + "the failed replication server must not be left registered"); + } + + // The listen port is never bound when the changelog cannot be read, and the aborted + // initialization leaves nothing holding it. + try (ServerSocket socket = new ServerSocket()) + { + socket.bind(new InetSocketAddress(ports[0])); + } + } + finally + { + // The aborted instance is never handed to the test, so its changelog cannot be removed + // through ReplicationTestCase.remove(). + recursiveDelete(dbDirectory); + } + } + /** Returns the names of the virtual attributes provided by the external changelog. */ private List changelogVirtualAttributeNames() { From ec16a29d5bcdbab8b11a9835b067464f0e038ad1 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 31 Jul 2026 20:11:08 +0300 Subject: [PATCH 2/5] [#802] Address review feedback on the fail-fast changelog read Reading the changelog restores one ReplicationServerDomain per domain it holds, and each of them starts its assured timer thread and registers its monitor provider. Every failure of that reading happens after that first loop, and so does a listen port which cannot be bound: abortInitialization() now shuts those domains down, as shutdown() does. The monitor instance name of a domain, and of its changelog, embeds the URL of its replication server, which was assigned only after the changelog had been read: the restored domains registered under a name holding a null URL, which no later lookup could match, so they leaked on the normal shutdown path too. setServerURL() now runs first. The listen thread owns the socket it was started on instead of reaching for the current one through a shared stopListen flag, so a port change starts the new listener before it stops the previous one. An interrupted wait for the previous listen thread can no longer leave the replication server with no listener at all. --- .../replication/server/ReplicationServer.java | 99 +++--- .../server/ReplicationServerListenThread.java | 14 +- .../changelog/file/FileChangelogDB.java | 9 +- .../ReplicationServerDynamicConfTest.java | 303 ++++++++++++++++++ 4 files changed, 377 insertions(+), 48 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index 14e8fb11d4..b23f67972b 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 @@ -130,8 +130,6 @@ public class ReplicationServer private boolean externalChangelogRegistered; private final AtomicBoolean shutdown = new AtomicBoolean(); - /** 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(); @@ -260,15 +258,22 @@ public static List getAllInstances() * This thread accept incoming connections on the replication server * ports from other replication servers or from LDAP servers * and spawn further thread responsible for handling those connections + *

+ * The socket is the one this thread was created for, not the one this replication server + * currently listens on: closing it is what stops this thread, and it is then the only + * thread it stops, even when another one is already listening on another port. + * + * @param socket + * the bound socket this thread accepts connections on */ - void runListen() + void runListen(ServerSocket socket) { logger.info(NOTE_REPLICATION_SERVER_LISTENING, getServerId(), - listenSocket.getInetAddress().getHostAddress(), - listenSocket.getLocalPort()); + socket.getInetAddress().getHostAddress(), + socket.getLocalPort()); - while (!shutdown.get() && !stopListen) + while (!shutdown.get() && !socket.isClosed()) { // Wait on the replicationServer port. // Read incoming messages and create LDAP or ReplicationServer listener @@ -279,7 +284,7 @@ void runListen() Socket newSocket = null; try { - newSocket = listenSocket.accept(); + newSocket = socket.accept(); newSocket.setTcpNoDelay(true); newSocket.setKeepAlive(true); int timeoutMS = MultimasterReplication.getConnectionTimeoutMS(); @@ -495,9 +500,13 @@ private void initialize() throws ConfigException try { + // Assigned before the changelog is opened: the monitor instance name of the domains it + // restores, and of their changelogs, embeds it, and a provider registered under a name + // which later changes can never be deregistered again. + setServerURL(); + this.changelogDB.initializeDB(); - setServerURL(); // Assigned before the threads are created, so that a failure below still releases it. listenSocket = bindListenPort(getReplicationPort()); @@ -523,8 +532,7 @@ private void initialize() throws ConfigException } catch (ChangelogException e) { // A replication server which cannot read its changelog is as dead as one which cannot - // bind its listen port: it would otherwise accept connections and fail on the first - // change it has to persist. The message already names the changelog directory. + // bind its listen port (issue #802). The message already names the changelog directory. logger.traceException(e); throw new ConfigException(e.getMessageObject(), e); } catch (UnknownHostException e) @@ -534,8 +542,8 @@ private void initialize() throws ConfigException 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. + // Every consumer would otherwise only learn about it as a "connection refused" + // somewhere else (issue #792). logger.traceException(e); throw new ConfigException(bindFailureMessage(getReplicationPort(), e), e); } @@ -589,28 +597,29 @@ private ServerSocket bindListenPort(int port) throws IOException } /** - * Stops the listen thread and releases the listen port. + * Stops the listen thread of the provided listen socket and releases that port. + *

+ * Both are the ones the thread was started on rather than the current ones, so this stops + * that thread only, even when another one is already listening on another port. * + * @param socket + * the listen socket to close, which is what stops its thread + * @param thread + * the listen thread of that socket, {@code null} when it was never started * @throws InterruptedException * if this thread is interrupted while waiting for the listen thread to stop */ - private void stopListenThread() throws InterruptedException + private void stopListenThread(ServerSocket socket, Thread thread) throws InterruptedException { - stopListen = true; - close(listenSocket); - if (listenThread != null) + close(socket); + if (thread != null) { - listenThread.join(); - listenThread = null; + thread.join(); } } /** * 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 @@ -618,18 +627,17 @@ private void stopListenThread() throws InterruptedException private void startListenThread(ServerSocket boundListenSocket) { listenSocket = boundListenSocket; - stopListen = false; - listenThread = new ReplicationServerListenThread(this); + listenThread = new ReplicationServerListenThread(this, boundListenSocket); 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. + * The new port is bound, and its listen thread started, while the current one is still + * open and serving. A failure therefore leaves this replication server listening on its + * current port, with its current configuration: there is nothing to roll back, and there + * is no window during which this replication server listens on no port at all. * * @param newConfig * the configuration being applied, whose listen port differs from the current one @@ -642,6 +650,8 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes { final ReplicationServerCfg previousConfig = this.config; final String previousServerURL = serverURL; + final ServerSocket previousListenSocket = listenSocket; + final Thread previousListenThread = listenThread; final int newPort = newConfig.getReplicationPort(); ServerSocket newListenSocket = null; try @@ -652,9 +662,21 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes this.config = newConfig; setServerURL(); - stopListenThread(); + // The new port is served before the current one is released, so that this replication + // server is never left with no listener at all, whatever happens next. startListenThread(newListenSocket); newListenSocket = null; + try + { + stopListenThread(previousListenSocket, previousListenThread); + } + catch (InterruptedException e) + { + // The previous port is already released and its thread stops on its own as soon as + // it wakes up on its closed socket: only the wait for it was cut short. + Thread.currentThread().interrupt(); + logger.traceException(e); + } localPorts.remove(previousConfig.getReplicationPort()); localPorts.add(newPort); @@ -673,15 +695,6 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes 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; @@ -761,6 +774,14 @@ private void abortInitialization() { listenThread.interrupt(); } + // Opening the changelog restores one domain per domain it holds, and each of them starts + // its threads and registers its monitor provider: a failure after that point, such as a + // listen port which cannot be bound, would otherwise leave them behind. Shut them down + // before the changelog they write to. + for (ReplicationServerDomain domain : getReplicationServerDomains()) + { + domain.shutdown(); + } shutdownExternalChangelog(); if (this.changelogDB != null) { diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java index f2a554be5e..0f91647dc7 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java @@ -13,9 +13,12 @@ * * Copyright 2008 Sun Microsystems, Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server; +import java.net.ServerSocket; + import org.opends.server.api.DirectoryThread; /** @@ -30,25 +33,30 @@ public class ReplicationServerListenThread extends DirectoryThread */ private final ReplicationServer server; + /** The socket this thread accepts connections on, and whose closing stops it. */ + private final ServerSocket listenSocket; + /** * Creates a new instance of this directory thread with the * specified name. * * @param server The ReplicationServer that will be called to * handle the connections. + * @param listenSocket The bound socket this thread will accept connections on. */ - public ReplicationServerListenThread(ReplicationServer server) + public ReplicationServerListenThread(ReplicationServer server, ServerSocket listenSocket) { super("Replication server RS(" + server.getServerId() + ") connection listener on port " - + server.getReplicationPort()); + + listenSocket.getLocalPort()); this.server = server; + this.listenSocket = listenSocket; } /** {@inheritDoc} */ @Override public void run() { - server.runListen(); + server.runListen(listenSocket); } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index ce6a9fdcbb..768d127e19 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -294,12 +294,9 @@ public void initializeDB() throws ChangelogException } catch (ChangelogException e) { - // A changelog which could not be read leaves this DB unusable: the replication - // environment may not exist at all, and the state which was restored before the failure - // only covers part of the domains. Both surface much later and somewhere else: as a - // failure on the first update to be persisted, or as a domain adopting the generation id - // of the first replica to connect, over a changelog which holds another generation. - // Not logged here: the caller reports the failure, logging it once. + // A changelog which could not be read leaves this DB unusable, and every shape of that + // failure surfaces much later and somewhere else (issue #802). Not logged here: the + // caller reports the failure, logging it once. logger.traceException(e); throw new ChangelogException( ERR_COULD_NOT_READ_DB.get(this.dbDirectory.getAbsolutePath(), e.getLocalizedMessage()), e); 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 52e3b9c2fd..b125f4a072 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 @@ -40,6 +40,8 @@ import org.opends.server.backends.ChangelogBackend; import org.opends.server.core.DirectoryServer; import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.server.changelog.api.ChangelogException; import org.opends.server.replication.service.ReplicationBroker; import org.opends.server.types.VirtualAttributeRule; @@ -317,6 +319,307 @@ public void replServerFailsWhenChangelogCannotBeRead() throws Exception } } + /** + * Tests the failure shape where the changelog state is restored only partially: the + * domains processed before the failure got their generation id and the ones after it did + * not, so they would adopt the generation id of the first replica to connect, over + * on-disk logs which belong to another generation. + *

+ * It is also the shape which restores domains before it fails, so it is the one where the + * aborted initialization has something to release. + */ + @Test + public void replServerFailsWhenAReplicaChangelogCannotBeRead() throws Exception + { + TestCaseUtils.startServer(); + + final int rsServerId = 8021; + final String dbDirName = "replServerFailsWhenAReplicaChangelogCannotBeReadDb"; + final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + try + { + // The head log file of the replica is replaced by a directory: the changelog state is + // then still readable, and the changes of the domain it names are not. + final File headLogFile = findFile(dbDirectory, "head", ".log"); + assertNotNull(headLogFile, "no replica changelog was written under " + dbDirectory); + assertTrue(headLogFile.delete() && headLogFile.mkdir(), "could not replace " + headLogFile); + + final int[] ports = TestCaseUtils.findFreePorts(1); + final int instancesBefore = ReplicationServer.getAllInstances().size(); + try + { + final ReplicationServer replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, rsServerId, 0, 0, null)); + remove(replicationServer); + fail("Creating a replication server over an unreadable replica changelog should have failed"); + } + catch (ConfigException expected) + { + assertTrue(expected.getCause() instanceof ChangelogException, + "the failure should be the one of the changelog, but was: " + expected.getCause()); + assertTrue(expected.getMessage().contains(dbDirectory.getAbsolutePath()), + "the failure should name the changelog directory, but was: " + expected.getMessage()); + assertEquals(ReplicationServer.getAllInstances().size(), instancesBefore, + "the failed replication server must not be left registered"); + assertNothingLeftBehind(rsServerId); + } + } + finally + { + recursiveDelete(dbDirectory); + } + } + + /** + * Tests that a replication server which cannot bind its listen port over a changelog it + * did read releases the domains that reading restored: each of them holds a timer thread + * and registers monitor providers, and the aborted instance is never handed to anything + * which could shut them down later. + */ + @Test + public void abortedStartReleasesTheRestoredDomains() throws Exception + { + TestCaseUtils.startServer(); + + final int rsServerId = 8022; + final String dbDirName = "abortedStartReleasesTheRestoredDomainsDb"; + final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) + { + try + { + final ReplicationServer replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(portHolder.getLocalPort(), dbDirName, 0, rsServerId, 0, 0, null)); + remove(replicationServer); + fail("Creating a replication server on a port already in use should have failed"); + } + catch (ConfigException expected) + { + assertNothingLeftBehind(rsServerId); + } + } + finally + { + recursiveDelete(dbDirectory); + } + } + + /** + * Tests that a replication server which restarted over an existing changelog releases the + * domains that reading it restored when it stops: their monitor instance name embeds the + * URL of their replication server, which used to be assigned only after the changelog had + * been read, so they were registered under a name holding a null URL and the name looked + * up to deregister them, built from the assigned URL, could never match it again. + */ + @Test + public void restartedReplServerReleasesTheRestoredDomains() throws Exception + { + TestCaseUtils.startServer(); + + final int rsServerId = 8023; + final String dbDirName = "restartedReplServerReleasesTheRestoredDomainsDb"; + final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + ReplicationServer replicationServer = null; + try + { + final int[] ports = TestCaseUtils.findFreePorts(1); + replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, rsServerId, 0, 0, null)); + assertTrue(replicationServer.isListening()); + assertFalse(domainRegistrationsOf(rsServerId).isEmpty(), + "the restored domain should hold a timer thread and monitor providers, otherwise" + + " this test does not test that they are released"); + } + finally + { + remove(replicationServer); + recursiveDelete(dbDirectory); + } + assertNothingLeftBehind(rsServerId); + } + + /** + * Tests that a port change whose wait for the previous listen thread is interrupted still + * leaves this replication server listening: the new port is served before the previous one + * is released, so an interrupt can only cut the wait for a thread which is already stopping + * short, never leave the replication server with no listener at all. + */ + @Test + public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + try + { + final int[] ports = TestCaseUtils.findFreePorts(2); + final String dbDirName = "replServerKeepsListeningWhenAPortChangeIsInterruptedDb"; + replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, 1, 0, 0, null)); + assertTrue(replicationServer.isListening()); + + // Thread.join() throws InterruptedException at once when the interrupt status is + // already set, i.e. this interrupts the port change in its wait for the listen thread. + Thread.currentThread().interrupt(); + final ConfigChangeResult ccr = replicationServer.applyConfigurationChange( + new ReplServerFakeConfiguration(ports[1], dbDirName, 0, 1, 0, 0, null)); + // Cleared for the rest of this test, and for whatever runs next in this thread. + final boolean interrupted = Thread.interrupted(); + + assertEquals(ccr.getResultCode(), ResultCode.SUCCESS); + assertTrue(interrupted, "the interrupted port change should have restored the interrupt status"); + assertEquals(replicationServer.getReplicationPort(), ports[1], + "the replication server should have switched to the new listen port"); + assertTrue(replicationServer.isListening(), + "an interrupted port change must not leave the replication server without a listener"); + + // and it must be usable on the new port. + ReplicationBroker broker = openReplicationSession( + DN.valueOf(TEST_ROOT_DN_STRING), 1, 10, ports[1], 1000); + assertTrue(broker.getCurrentSendWindow() != 0); + } + finally + { + remove(replicationServer); + } + } + + /** + * Creates the changelog of a replication server which ran and served one replica, and + * returns its directory: a changelog whose reading restores a domain, i.e. one over which + * an initialization has something to release when it fails. + */ + private File createPopulatedChangelog(String dbDirName, int rsServerId) throws Exception + { + final File dbDirectory = getFileForPath(dbDirName); + recursiveDelete(dbDirectory); + + final int[] ports = TestCaseUtils.findFreePorts(1); + final ReplicationServer replicationServer = new ReplicationServer( + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, rsServerId, 0, 0, null)); + ReplicationBroker broker = null; + try + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + broker = openReplicationSession(baseDN, 42, 100, ports[0], 1000); + broker.publish(new DeleteMsg(baseDN, new CSNGenerator(42, 0).newCSN(), "uid")); + + // The changelog is written by the replication server, so the publication above is only + // over once the log of the replica exists. + waitFor(dbDirectory, "head", ".log"); + } + finally + { + stop(broker); + replicationServer.shutdown(); + } + + assertNotNull(findFile(dbDirectory, "generation", ".id"), + "the changelog should hold the generation id of the domain, otherwise reading it back" + + " restores no domain at all and this test tests nothing"); + // A replication server which stopped normally must not leave anything behind either. + assertNothingLeftBehind(rsServerId); + return dbDirectory; + } + + /** + * Asserts that the replication server with the provided server id left neither a thread of + * a domain nor a monitor provider of a domain or of its changelog behind. + */ + private void assertNothingLeftBehind(int rsServerId) throws Exception + { + // Stopping a thread only asks it to stop, so give the ones being stopped the time to + // actually stop before reporting them as left behind. + final long deadline = System.currentTimeMillis() + 10000; + List leftBehind; + while (!(leftBehind = domainRegistrationsOf(rsServerId)).isEmpty() && System.currentTimeMillis() < deadline) + { + Thread.sleep(10); + } + assertEquals(leftBehind, Collections.emptyList(), + "the replication server RS(" + rsServerId + ") left the above behind"); + } + + /** The threads a {@link ReplicationServerDomain} starts, and which its shutdown stops. */ + private static final Collection DOMAIN_THREADS = + Arrays.asList("assured timer for domain", "status monitor for domain"); + + /** + * Returns the threads and the monitor providers which the domains of the replication server + * with the provided server id have started and registered. + */ + private List domainRegistrationsOf(int rsServerId) + { + final String replicationServer = "replication server rs(" + rsServerId + ")"; + final List registrations = new ArrayList<>(); + for (Thread thread : Thread.getAllStackTraces().keySet()) + { + final String name = thread.getName().toLowerCase(); + if (name.startsWith(replicationServer) && containsAnyOf(name, DOMAIN_THREADS)) + { + registrations.add(thread.getName()); + } + } + // The monitor instance names are registered in lowercase. + for (String monitorName : DirectoryServer.getMonitorProviders().keySet()) + { + if (monitorName.contains(replicationServer)) + { + registrations.add(monitorName); + } + } + Collections.sort(registrations); + return registrations; + } + + private boolean containsAnyOf(String name, Collection candidates) + { + for (String candidate : candidates) + { + if (name.contains(candidate)) + { + return true; + } + } + return false; + } + + /** Waits for a file whose name matches to appear anywhere under the provided directory. */ + private void waitFor(File directory, String prefix, String suffix) throws Exception + { + final long deadline = System.currentTimeMillis() + 10000; + while (findFile(directory, prefix, suffix) == null && System.currentTimeMillis() < deadline) + { + Thread.sleep(10); + } + assertNotNull(findFile(directory, prefix, suffix), + "no " + prefix + "*" + suffix + " was written under " + directory); + } + + /** Returns the first file whose name matches, at any depth of the provided directory. */ + private File findFile(File directory, String prefix, String suffix) + { + final File[] files = directory.listFiles(); + if (files == null) + { + return null; + } + for (File file : files) + { + final String name = file.getName(); + if (name.startsWith(prefix) && name.endsWith(suffix)) + { + return file; + } + final File found = file.isDirectory() ? findFile(file, prefix, suffix) : null; + if (found != null) + { + return found; + } + } + return null; + } + /** Returns the names of the virtual attributes provided by the external changelog. */ private List changelogVirtualAttributeNames() { From fc8337d1a4e809b7261641d874903e8242e1201f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 3 Aug 2026 12:51:32 +0300 Subject: [PATCH 3/5] [#802] Address round-5 review feedback on the fail-fast changelog read The changelog of the change number index holds a head log file of its own, created when the replication server starts, so the unscoped lookup in the test helper could match it instead of the log of the replica: the wait then waited for nothing and the corruption hit the wrong log. The lookup is now scoped to the domain directories, and the test asserts that the failure names the replica log it corrupted, so it cannot pass while exercising another failure shape. A port change whose wait for the previous listen thread is interrupted is now exercised rather than passed over: Thread.join() only throws while the thread it waits for is alive, so the test keeps that thread in its handshake with a connection which says nothing, and asserts the counter which only the interrupted path increments. abortInitialization() shuts each restored domain down on its own, so that an unchecked failure of one of them does not skip the changelog shutdown which follows. --- .../replication/server/ReplicationServer.java | 26 ++- .../ReplicationServerDynamicConfTest.java | 163 ++++++++++++++---- 2 files changed, 156 insertions(+), 33 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index b23f67972b..0f957a208f 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 @@ -153,6 +153,15 @@ public class ReplicationServer */ static final AtomicInteger listenPortBindFailures = new AtomicInteger(); + /** + * Number of listen port changes whose wait for the previous listen thread was interrupted. + *

+ * This is required for unit testing: a wait which is interrupted and a wait which is over + * before it starts leave this replication server in the same state, so nothing else tells + * a test that it exercised the interruption instead of passing over it. + */ + static final AtomicInteger interruptedListenThreadStops = new AtomicInteger(); + /** Monitors for synchronizing domain creation with the connect thread. */ private final Object domainTicketLock = new Object(); private final Object connectThreadLock = new Object(); @@ -638,6 +647,10 @@ private void startListenThread(ServerSocket boundListenSocket) * open and serving. A failure therefore leaves this replication server listening on its * current port, with its current configuration: there is nothing to roll back, and there * is no window during which this replication server listens on no port at all. + *

+ * The trade is a window during which both ports accept, so a peer which connects to the + * previous port just before it is released gets a session which outlives the change. That + * is the deliberate inverse of a window during which nothing listens at all. * * @param newConfig * the configuration being applied, whose listen port differs from the current one @@ -674,6 +687,7 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes { // The previous port is already released and its thread stops on its own as soon as // it wakes up on its closed socket: only the wait for it was cut short. + interruptedListenThreadStops.incrementAndGet(); Thread.currentThread().interrupt(); logger.traceException(e); } @@ -777,10 +791,18 @@ private void abortInitialization() // Opening the changelog restores one domain per domain it holds, and each of them starts // its threads and registers its monitor provider: a failure after that point, such as a // listen port which cannot be bound, would otherwise leave them behind. Shut them down - // before the changelog they write to. + // before the changelog they write to, and one failure at a time: the changelog this one + // is built on is known to be broken, and what follows still has to run. for (ReplicationServerDomain domain : getReplicationServerDomains()) { - domain.shutdown(); + try + { + domain.shutdown(); + } + catch (RuntimeException ignored) + { + logger.traceException(ignored); + } } shutdownExternalChangelog(); if (this.changelogDB != null) 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 b125f4a072..720f857ec3 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 @@ -24,6 +24,7 @@ import java.io.File; import java.net.InetSocketAddress; import java.net.ServerSocket; +import java.net.Socket; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; @@ -31,6 +32,7 @@ import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Map; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.opendj.config.server.ConfigChangeResult; @@ -335,12 +337,12 @@ public void replServerFailsWhenAReplicaChangelogCannotBeRead() throws Exception final int rsServerId = 8021; final String dbDirName = "replServerFailsWhenAReplicaChangelogCannotBeReadDb"; - final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + final File dbDirectory = createPopulatedChangelog(dbDirName); try { // The head log file of the replica is replaced by a directory: the changelog state is // then still readable, and the changes of the domain it names are not. - final File headLogFile = findFile(dbDirectory, "head", ".log"); + final File headLogFile = findReplicaLogFile(dbDirectory); assertNotNull(headLogFile, "no replica changelog was written under " + dbDirectory); assertTrue(headLogFile.delete() && headLogFile.mkdir(), "could not replace " + headLogFile); @@ -359,6 +361,11 @@ public void replServerFailsWhenAReplicaChangelogCannotBeRead() throws Exception "the failure should be the one of the changelog, but was: " + expected.getCause()); assertTrue(expected.getMessage().contains(dbDirectory.getAbsolutePath()), "the failure should name the changelog directory, but was: " + expected.getMessage()); + // The log of the replica, named after its directory, and not the one of the change + // number index: otherwise this test exercises another failure shape than its name. + assertTrue(expected.getMessage().contains(headLogFile.getParentFile().getPath()), + "the failure should be the one of the replica changelog which was corrupted," + + " but was: " + expected.getMessage()); assertEquals(ReplicationServer.getAllInstances().size(), instancesBefore, "the failed replication server must not be left registered"); assertNothingLeftBehind(rsServerId); @@ -383,7 +390,7 @@ public void abortedStartReleasesTheRestoredDomains() throws Exception final int rsServerId = 8022; final String dbDirName = "abortedStartReleasesTheRestoredDomainsDb"; - final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + final File dbDirectory = createPopulatedChangelog(dbDirName); try (ServerSocket portHolder = TestCaseUtils.bindFreePort()) { try @@ -418,7 +425,7 @@ public void restartedReplServerReleasesTheRestoredDomains() throws Exception final int rsServerId = 8023; final String dbDirName = "restartedReplServerReleasesTheRestoredDomainsDb"; - final File dbDirectory = createPopulatedChangelog(dbDirName, rsServerId); + final File dbDirectory = createPopulatedChangelog(dbDirName); ReplicationServer replicationServer = null; try { @@ -441,8 +448,14 @@ public void restartedReplServerReleasesTheRestoredDomains() throws Exception /** * Tests that a port change whose wait for the previous listen thread is interrupted still * leaves this replication server listening: the new port is served before the previous one - * is released, so an interrupt can only cut the wait for a thread which is already stopping - * short, never leave the replication server with no listener at all. + * is released, so an interrupt can only cut short the wait for a thread which is already + * stopping, never leave the replication server with no listener at all. + *

+ * A connection which is accepted and then says nothing keeps the previous listen thread in + * its handshake instead of at {@code accept()}, so closing its socket does not stop it + * before the wait even begins: {@code Thread.join()} only throws while the thread it waits + * for is alive, so without that connection this test would pass over the interruption + * instead of exercising it. {@code interruptedListenThreadStops} tells the two apart. */ @Test public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Exception @@ -458,16 +471,27 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except new ReplServerFakeConfiguration(ports[0], dbDirName, 0, 1, 0, 0, null)); assertTrue(replicationServer.isListening()); - // Thread.join() throws InterruptedException at once when the interrupt status is - // already set, i.e. this interrupts the port change in its wait for the listen thread. - Thread.currentThread().interrupt(); - final ConfigChangeResult ccr = replicationServer.applyConfigurationChange( - new ReplServerFakeConfiguration(ports[1], dbDirName, 0, 1, 0, 0, null)); + final int interruptsBefore = ReplicationServer.interruptedListenThreadStops.get(); + final ConfigChangeResult ccr; + try (Socket silent = new Socket()) + { + silent.connect(new InetSocketAddress("127.0.0.1", ports[0]), 5000); + waitForListenThreadOf(ports[0]); + + // Thread.join() throws InterruptedException at once when the interrupt status is + // already set, i.e. this interrupts the port change in its wait for the listen thread. + Thread.currentThread().interrupt(); + ccr = replicationServer.applyConfigurationChange( + new ReplServerFakeConfiguration(ports[1], dbDirName, 0, 1, 0, 0, null)); + } // Cleared for the rest of this test, and for whatever runs next in this thread. final boolean interrupted = Thread.interrupted(); - assertEquals(ccr.getResultCode(), ResultCode.SUCCESS); + assertEquals(ReplicationServer.interruptedListenThreadStops.get(), interruptsBefore + 1, + "the port change should have been interrupted in its wait for the previous listen" + + " thread, otherwise this test does not test that interruption"); assertTrue(interrupted, "the interrupted port change should have restored the interrupt status"); + assertEquals(ccr.getResultCode(), ResultCode.SUCCESS); assertEquals(replicationServer.getReplicationPort(), ports[1], "the replication server should have switched to the new listen port"); assertTrue(replicationServer.isListening(), @@ -484,19 +508,67 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except } } + /** + * Waits for the listen thread of the provided port to leave {@code accept()}, i.e. for the + * connection which was just made to it to have been accepted. + */ + private void waitForListenThreadOf(int port) throws Exception + { + final String listenThread = "replication server rs(1) connection listener on port " + port; + final long deadline = System.currentTimeMillis() + 60000; + while (System.currentTimeMillis() < deadline) + { + for (Map.Entry entry : Thread.getAllStackTraces().entrySet()) + { + if (entry.getKey().getName().toLowerCase().equals(listenThread) + && !isAccepting(entry.getValue())) + { + return; + } + } + Thread.sleep(10); + } + fail("the listen thread on port " + port + " never accepted the connection made to it"); + } + + private boolean isAccepting(StackTraceElement[] stackTrace) + { + for (StackTraceElement frame : stackTrace) + { + if ("java.net.ServerSocket".equals(frame.getClassName()) && frame.getMethodName().contains("accept")) + { + return true; + } + } + return false; + } + + /** + * The server id of the replication server which writes the changelog the tests read back. + *

+ * It is not the one of the replication servers under test, so that what it leaves behind — + * it is the only one to which a replica ever connects, and the monitor providers of a + * connection are deregistered when its handler notices it is gone — cannot be mistaken for + * what they leave behind. + */ + private static final int CHANGELOG_WRITER_RS_ID = 8020; + + /** The suffix of the directory a changelog holds per domain, see {@code ReplicationEnvironment}. */ + private static final String DOMAIN_DIRECTORY_SUFFIX = ".dom"; + /** * Creates the changelog of a replication server which ran and served one replica, and * returns its directory: a changelog whose reading restores a domain, i.e. one over which * an initialization has something to release when it fails. */ - private File createPopulatedChangelog(String dbDirName, int rsServerId) throws Exception + private File createPopulatedChangelog(String dbDirName) throws Exception { final File dbDirectory = getFileForPath(dbDirName); recursiveDelete(dbDirectory); final int[] ports = TestCaseUtils.findFreePorts(1); final ReplicationServer replicationServer = new ReplicationServer( - new ReplServerFakeConfiguration(ports[0], dbDirName, 0, rsServerId, 0, 0, null)); + new ReplServerFakeConfiguration(ports[0], dbDirName, 0, CHANGELOG_WRITER_RS_ID, 0, 0, null)); ReplicationBroker broker = null; try { @@ -504,21 +576,17 @@ private File createPopulatedChangelog(String dbDirName, int rsServerId) throws E broker = openReplicationSession(baseDN, 42, 100, ports[0], 1000); broker.publish(new DeleteMsg(baseDN, new CSNGenerator(42, 0).newCSN(), "uid")); - // The changelog is written by the replication server, so the publication above is only - // over once the log of the replica exists. - waitFor(dbDirectory, "head", ".log"); + // The changelog is written by the replication server, so the calls above are only over + // once the log of the replica exists. It is the last of the four files which + // ReplicationEnvironment.getOrCreateReplicaDB() writes, after domains.state, the server + // id directory and the generation id, so waiting for it waits for all of them. + waitForReplicaLogFile(dbDirectory); } finally { stop(broker); replicationServer.shutdown(); } - - assertNotNull(findFile(dbDirectory, "generation", ".id"), - "the changelog should hold the generation id of the domain, otherwise reading it back" - + " restores no domain at all and this test tests nothing"); - // A replication server which stopped normally must not leave anything behind either. - assertNothingLeftBehind(rsServerId); return dbDirectory; } @@ -536,8 +604,8 @@ private void assertNothingLeftBehind(int rsServerId) throws Exception { Thread.sleep(10); } - assertEquals(leftBehind, Collections.emptyList(), - "the replication server RS(" + rsServerId + ") left the above behind"); + assertTrue(leftBehind.isEmpty(), + "the replication server RS(" + rsServerId + ") left behind: " + leftBehind); } /** The threads a {@link ReplicationServerDomain} starts, and which its shutdown stops. */ @@ -584,16 +652,49 @@ private boolean containsAnyOf(String name, Collection candidates) return false; } - /** Waits for a file whose name matches to appear anywhere under the provided directory. */ - private void waitFor(File directory, String prefix, String suffix) throws Exception + /** + * Waits for the head log file of a replica changelog to appear under the provided changelog. + *

+ * The wait is long because it is only ever reached on a machine which is slow enough for + * the replication server to still be writing that file: a test which passes never waits. + */ + private void waitForReplicaLogFile(File dbDirectory) throws Exception { - final long deadline = System.currentTimeMillis() + 10000; - while (findFile(directory, prefix, suffix) == null && System.currentTimeMillis() < deadline) + final long deadline = System.currentTimeMillis() + 60000; + while (findReplicaLogFile(dbDirectory) == null && System.currentTimeMillis() < deadline) { Thread.sleep(10); } - assertNotNull(findFile(directory, prefix, suffix), - "no " + prefix + "*" + suffix + " was written under " + directory); + assertNotNull(findReplicaLogFile(dbDirectory), + "no replica changelog was written under " + dbDirectory); + } + + /** + * Returns the head log file of a replica changelog, i.e. the one under a domain directory. + *

+ * The changelog of the change number index holds a head log file of its own, and it is + * created when the replication server starts: a lookup which is not scoped to a domain + * directory can match it instead, and then waits for nothing and corrupts the wrong log. + */ + private File findReplicaLogFile(File dbDirectory) + { + final File[] entries = dbDirectory.listFiles(); + if (entries == null) + { + return null; + } + for (File entry : entries) + { + if (entry.isDirectory() && entry.getName().endsWith(DOMAIN_DIRECTORY_SUFFIX)) + { + final File replicaLogFile = findFile(entry, "head", ".log"); + if (replicaLogFile != null) + { + return replicaLogFile; + } + } + } + return null; } /** Returns the first file whose name matches, at any depth of the provided directory. */ From fe88515f40cef7599008bb46b38528d91e2e3c71 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 3 Aug 2026 14:31:28 +0300 Subject: [PATCH 4/5] [#802] Address round-6 review feedback on the fail-fast changelog read The wait for the listen thread of the interrupted port change returned as soon as that thread was seen outside accept(), which a thread that has not reached accept() yet also satisfies: a connection completes against the listen backlog of the kernel, and isListening() only reports a bound socket, so the connection could still be queued when the port change closed the socket under it. The previous listen thread then died before the wait for it began, and the test failed on its own interruptedListenThreadStops assertion. The helper now waits for that thread to be inside accept() before the connection is made and to have left it afterwards, which it can only have done for that connection, and it takes the server id instead of hardcoding rs(1). listenThread becomes volatile, like the listen socket it goes with: a port change reads it from the configuration thread, and a stale read would skip the join it was captured for. The comment of abortInitialization() no longer promises more than its catch of RuntimeException delivers, and the wait for a replica changelog names the domain directory suffix it looks for, so that a rename of the constant which owns it reads as such. --- .../replication/server/ReplicationServer.java | 7 +-- .../ReplicationServerDynamicConfTest.java | 47 +++++++++++++++---- 2 files changed, 43 insertions(+), 11 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index 0f957a208f..c9877c47bc 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 @@ -106,7 +106,8 @@ public class ReplicationServer private static final int LISTEN_PORT_PROBE_TIMEOUT_MS = 200; private volatile ServerSocket listenSocket; - private Thread listenThread; + /** Volatile like its socket above: a port change reads it from the configuration thread. */ + private volatile Thread listenThread; private Thread connectThread; /** The current configuration of this replication server. */ @@ -791,8 +792,8 @@ private void abortInitialization() // Opening the changelog restores one domain per domain it holds, and each of them starts // its threads and registers its monitor provider: a failure after that point, such as a // listen port which cannot be bound, would otherwise leave them behind. Shut them down - // before the changelog they write to, and one failure at a time: the changelog this one - // is built on is known to be broken, and what follows still has to run. + // before the changelog they write to, and one unchecked exception at a time: the changelog + // this one is built on is known to be broken, and what follows still has to run. for (ReplicationServerDomain domain : getReplicationServerDomains()) { try 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 720f857ec3..abb6f285ec 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 @@ -456,6 +456,9 @@ public void restartedReplServerReleasesTheRestoredDomains() throws Exception * before the wait even begins: {@code Thread.join()} only throws while the thread it waits * for is alive, so without that connection this test would pass over the interruption * instead of exercising it. {@code interruptedListenThreadStops} tells the two apart. + *

+ * That the connection is accepted is waited for on both sides of it, see + * {@link #waitForListenThread(int, int, boolean)}: connecting only proves the port is bound. */ @Test public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Exception @@ -475,11 +478,20 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except final ConfigChangeResult ccr; try (Socket silent = new Socket()) { + // The listen thread has to be inside accept() before the connection is made, and to + // have left it afterwards: a connection completes against the listen backlog of the + // kernel, so it does not prove that the thread which serves it ever ran. + final int serverId = replicationServer.getServerId(); + waitForListenThread(serverId, ports[0], true); silent.connect(new InetSocketAddress("127.0.0.1", ports[0]), 5000); - waitForListenThreadOf(ports[0]); + waitForListenThread(serverId, ports[0], false); // Thread.join() throws InterruptedException at once when the interrupt status is // already set, i.e. this interrupts the port change in its wait for the listen thread. + // What runs until that wait — binding the new port, resolving the server URL, starting + // the new listen thread — has to fit in the handshake timeout of the silent connection, + // MultimasterReplication.getConnectionTimeoutMS(), 5s by default: that handshake is + // what keeps the previous listen thread alive, hence what makes the wait for it block. Thread.currentThread().interrupt(); ccr = replicationServer.applyConfigurationChange( new ReplServerFakeConfiguration(ports[1], dbDirName, 0, 1, 0, 0, null)); @@ -509,26 +521,43 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except } /** - * Waits for the listen thread of the provided port to leave {@code accept()}, i.e. for the - * connection which was just made to it to have been accepted. + * Waits for the listen thread of the provided replication server and port to be inside + * {@code accept()}, or to have left it. + *

+ * Having left {@code accept()} is what tells that the connection which was just made to + * that port is being served: a connection completes against the listen backlog of the + * kernel, and {@code isListening()} only tells that the socket is bound — the listen + * thread is started after it — so neither of them proves that thread ever ran. Waiting for + * it to be inside {@code accept()} first is what makes the second wait conclusive: it can + * then only have left it for the connection this test made. + * + * @param serverId + * the server id of the replication server whose listen thread is waited for + * @param port + * the port that listen thread listens on + * @param accepting + * {@code true} to wait for that thread to be inside {@code accept()}, + * {@code false} to wait for it to have left it */ - private void waitForListenThreadOf(int port) throws Exception + private void waitForListenThread(int serverId, int port, boolean accepting) throws Exception { - final String listenThread = "replication server rs(1) connection listener on port " + port; + final String listenThread = + "replication server rs(" + serverId + ") connection listener on port " + port; final long deadline = System.currentTimeMillis() + 60000; while (System.currentTimeMillis() < deadline) { for (Map.Entry entry : Thread.getAllStackTraces().entrySet()) { if (entry.getKey().getName().toLowerCase().equals(listenThread) - && !isAccepting(entry.getValue())) + && isAccepting(entry.getValue()) == accepting) { return; } } Thread.sleep(10); } - fail("the listen thread on port " + port + " never accepted the connection made to it"); + fail("the listen thread on port " + port + + (accepting ? " never reached accept()" : " never accepted the connection made to it")); } private boolean isAccepting(StackTraceElement[] stackTrace) @@ -666,7 +695,9 @@ private void waitForReplicaLogFile(File dbDirectory) throws Exception Thread.sleep(10); } assertNotNull(findReplicaLogFile(dbDirectory), - "no replica changelog was written under " + dbDirectory); + "no replica changelog, i.e. no head log file under a '" + DOMAIN_DIRECTORY_SUFFIX + + "' directory, was written under " + dbDirectory + + ": check that suffix against ReplicationEnvironment, which owns it"); } /** From c3b13490292a71110c34700091115dfddca2b6be Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 3 Aug 2026 17:27:13 +0300 Subject: [PATCH 5/5] [#802] Address round-7 review feedback on the fail-fast changelog read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The interrupt of a cut-short wait for the previous listen thread escaped into the rest of applyConfigurationChange(): some of what follows the port switch is interruptible — stopping the handlers of removed replication servers takes the domain lock interruptibly, after a one-shot engageShutdown() — so a change combining a port change with a server list change could leave a handler engaged but not stopped while reporting SUCCESS. switchListenPort() now records the interrupt in a holder its caller passes, and applyConfigurationChange() restores it once the rest of the change has run: a parameter rather than a field, so nothing survives the call. localPorts is updated as soon as the new listen thread serves, instead of trailing getReplicationPort() for as long as the wait for the previous thread blocks. In the test, the two waits built on thread dumps poll every 50ms instead of every 10ms, so that their failure path is not dominated by the dumps it takes, the thread names they match with equals() are lowercased with Locale.ROOT, and the javadoc of waitForListenThread() states the invariant it establishes — the thread is off accept() and cannot terminate until its socket is closed — instead of claiming the connection is served, along with the assumption that this connection is the only one its port ever gets. --- .../replication/server/ReplicationServer.java | 30 ++++++++++++++---- .../ReplicationServerDynamicConfTest.java | 31 +++++++++++++------ 2 files changed, 45 insertions(+), 16 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index c9877c47bc..4bea16769d 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 @@ -657,10 +657,15 @@ private void startListenThread(ServerSocket boundListenSocket) * 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 + * @param listenThreadStopInterrupted + * set when the wait for the previous listen thread was interrupted, instead of + * restoring the interrupt status here: the caller restores it once the rest of + * the change, some of which is interruptible, has run * @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) + private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeResult ccr, + AtomicBoolean listenThreadStopInterrupted) { final ReplicationServerCfg previousConfig = this.config; final String previousServerURL = serverURL; @@ -680,6 +685,12 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes // server is never left with no listener at all, whatever happens next. startListenThread(newListenSocket); newListenSocket = null; + + // In step with getReplicationPort(), which answers the new port from here on: the + // wait below blocks for as long as the previous thread takes to serve its current + // connection, and localPorts must not trail it for that whole window. + localPorts.remove(previousConfig.getReplicationPort()); + localPorts.add(newPort); try { stopListenThread(previousListenSocket, previousListenThread); @@ -689,12 +700,9 @@ private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeRes // The previous port is already released and its thread stops on its own as soon as // it wakes up on its closed socket: only the wait for it was cut short. interruptedListenThreadStops.incrementAndGet(); - Thread.currentThread().interrupt(); + listenThreadStopInterrupted.set(true); logger.traceException(e); } - - localPorts.remove(previousConfig.getReplicationPort()); - localPorts.add(newPort); return true; } catch (UnknownHostException e) @@ -1246,8 +1254,9 @@ public ConfigChangeResult applyConfigurationChange( // 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. + final AtomicBoolean listenThreadStopInterrupted = new AtomicBoolean(); if (configuration.getReplicationPort() != oldConfig.getReplicationPort() - && !switchListenPort(configuration, ccr)) + && !switchListenPort(configuration, ccr, listenThreadStopInterrupted)) { return ccr; } @@ -1313,6 +1322,15 @@ public ConfigChangeResult applyConfigurationChange( { ccr.setAdminActionRequired(true); } + + // The interrupt which cut short the wait for the previous listen thread, deferred by + // switchListenPort(): restored only now, because the steps above include interruptible + // ones — stopping the handlers of removed replication servers locks interruptibly — + // which an interrupt status left set would have failed while the change reports SUCCESS. + if (listenThreadStopInterrupted.get()) + { + Thread.currentThread().interrupt(); + } return ccr; } 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 abb6f285ec..c15e5379e5 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 @@ -32,6 +32,7 @@ import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Locale; import java.util.Map; import org.forgerock.i18n.LocalizableMessage; @@ -524,12 +525,18 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except * Waits for the listen thread of the provided replication server and port to be inside * {@code accept()}, or to have left it. *

- * Having left {@code accept()} is what tells that the connection which was just made to - * that port is being served: a connection completes against the listen backlog of the - * kernel, and {@code isListening()} only tells that the socket is bound — the listen - * thread is started after it — so neither of them proves that thread ever ran. Waiting for - * it to be inside {@code accept()} first is what makes the second wait conclusive: it can - * then only have left it for the connection this test made. + * A connection completes against the listen backlog of the kernel, and + * {@code isListening()} only tells that the socket is bound — the listen thread is + * started after it — so neither of them proves that thread ever ran. Waiting for it to be + * inside {@code accept()} before connecting is what makes the second wait conclusive: the + * thread can then only have left {@code accept()} for the connection this test made. That + * rests on it being the only connection the port ever gets — a stale broker of an earlier + * test reconnecting to a recycled port would satisfy the second wait spuriously. + *

+ * Having left {@code accept()} does not mean the connection is served yet — the thread is + * typically still warming up towards its handshake. What the second wait establishes is + * that the thread is off {@code accept()} and cannot terminate until its socket is + * closed, which is what stopping it does. * * @param serverId * the server id of the replication server whose listen thread is waited for @@ -548,13 +555,15 @@ private void waitForListenThread(int serverId, int port, boolean accepting) thro { for (Map.Entry entry : Thread.getAllStackTraces().entrySet()) { - if (entry.getKey().getName().toLowerCase().equals(listenThread) + if (entry.getKey().getName().toLowerCase(Locale.ROOT).equals(listenThread) && isAccepting(entry.getValue()) == accepting) { return; } } - Thread.sleep(10); + // Each iteration is a full VM thread dump: poll slowly enough for the failure path + // not to be dominated by them, the passing case returns within an iteration or two. + Thread.sleep(50); } fail("the listen thread on port " + port + (accepting ? " never reached accept()" : " never accepted the connection made to it")); @@ -631,7 +640,9 @@ private void assertNothingLeftBehind(int rsServerId) throws Exception List leftBehind; while (!(leftBehind = domainRegistrationsOf(rsServerId)).isEmpty() && System.currentTimeMillis() < deadline) { - Thread.sleep(10); + // Each iteration is a full VM thread dump: poll slowly enough for the failure path + // not to be dominated by them, the passing case returns within an iteration or two. + Thread.sleep(50); } assertTrue(leftBehind.isEmpty(), "the replication server RS(" + rsServerId + ") left behind: " + leftBehind); @@ -651,7 +662,7 @@ private List domainRegistrationsOf(int rsServerId) final List registrations = new ArrayList<>(); for (Thread thread : Thread.getAllStackTraces().keySet()) { - final String name = thread.getName().toLowerCase(); + final String name = thread.getName().toLowerCase(Locale.ROOT); if (name.startsWith(replicationServer) && containsAnyOf(name, DOMAIN_THREADS)) { registrations.add(thread.getName());