From bed2b17a444ee960268da0c05d1ca5365bdf2d98 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 04 Aug 2026 06:29:59 +0000
Subject: [PATCH] [#802] Fail fast when the replication server cannot read its changelog (#805)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java | 153 +++++++++---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java | 8
opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java | 8
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java | 14
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java | 506 ++++++++++++++++++++++++++++++++++++++++++
5 files changed, 641 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 d94f169..4bea167 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 @@
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. */
@@ -130,8 +131,6 @@
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();
@@ -155,6 +154,15 @@
*/
static final AtomicInteger listenPortBindFailures = new AtomicInteger();
+ /**
+ * Number of listen port changes whose wait for the previous listen thread was interrupted.
+ * <p>
+ * 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();
@@ -260,15 +268,22 @@
* 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
+ * <p>
+ * 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 +294,7 @@
Socket newSocket = null;
try
{
- newSocket = listenSocket.accept();
+ newSocket = socket.accept();
newSocket.setTcpNoDelay(true);
newSocket.setKeepAlive(true);
int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
@@ -486,8 +501,8 @@
* 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
{
@@ -495,9 +510,13 @@
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());
@@ -520,6 +539,12 @@
{
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 (issue #802). 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.
@@ -527,8 +552,8 @@
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);
}
@@ -582,28 +607,29 @@
}
/**
- * Stops the listen thread and releases the listen port.
+ * Stops the listen thread of the provided listen socket and releases that port.
+ * <p>
+ * 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.
- * <p>
- * {@code stopListen} is only cleared here, i.e. once the listen port is bound, so a
- * failure to bind leaves this replication server consistently stopped rather than with a
- * listen thread which would spin on a closed socket.
*
* @param boundListenSocket
* the bound socket the listen thread will accept connections on
@@ -611,30 +637,40 @@
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.
* <p>
- * The new port is bound while the current one is still open and serving, so a failure
- * leaves this replication server listening on its current port, with its current
- * configuration: there is nothing to roll back, and no window during which this
- * replication server advertises a port that nothing listens to.
+ * 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.
+ * <p>
+ * 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
* @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;
+ final ServerSocket previousListenSocket = listenSocket;
+ final Thread previousListenThread = listenThread;
final int newPort = newConfig.getReplicationPort();
ServerSocket newListenSocket = null;
try
@@ -645,12 +681,28 @@
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;
+ // 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);
+ }
+ 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.
+ interruptedListenThreadStops.incrementAndGet();
+ listenThreadStopInterrupted.set(true);
+ logger.traceException(e);
+ }
return true;
}
catch (UnknownHostException e)
@@ -666,15 +718,6 @@
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;
@@ -754,6 +797,22 @@
{
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, 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
+ {
+ domain.shutdown();
+ }
+ catch (RuntimeException ignored)
+ {
+ logger.traceException(ignored);
+ }
+ }
shutdownExternalChangelog();
if (this.changelogDB != null)
{
@@ -1195,8 +1254,9 @@
// 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;
}
@@ -1262,6 +1322,15 @@
{
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/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerListenThread.java
index f2a554b..0f91647 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 @@
*/
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/api/ChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/api/ChangelogDB.java
index 9b9e3ef..a75f7da 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 @@
* 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 2164ae9..768d127 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 @@
}
@Override
- public void initializeDB()
+ public void initializeDB() throws ChangelogException
{
try
{
@@ -294,8 +294,12 @@
}
catch (ChangelogException e)
{
+ // 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);
- 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 ff20e85..c15e537 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,12 +21,19 @@
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.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
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;
import org.forgerock.opendj.config.server.ConfigChangeResult;
@@ -36,6 +43,9 @@
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;
import org.forgerock.opendj.ldap.DN;
@@ -257,6 +267,502 @@
}
}
+ /**
+ * 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);
+ }
+ }
+
+ /**
+ * 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.
+ * <p>
+ * 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);
+ 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 = findReplicaLogFile(dbDirectory);
+ 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());
+ // 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);
+ }
+ }
+ 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);
+ 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);
+ 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 short the wait for a thread which is already
+ * stopping, never leave the replication server with no listener at all.
+ * <p>
+ * 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.
+ * <p>
+ * 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
+ {
+ 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());
+
+ final int interruptsBefore = ReplicationServer.interruptedListenThreadStops.get();
+ 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);
+ 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));
+ }
+ // Cleared for the rest of this test, and for whatever runs next in this thread.
+ final boolean interrupted = Thread.interrupted();
+
+ 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(),
+ "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);
+ }
+ }
+
+ /**
+ * Waits for the listen thread of the provided replication server and port to be inside
+ * {@code accept()}, or to have left it.
+ * <p>
+ * 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.
+ * <p>
+ * 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
+ * @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 waitForListenThread(int serverId, int port, boolean accepting) throws Exception
+ {
+ 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<Thread, StackTraceElement[]> entry : Thread.getAllStackTraces().entrySet())
+ {
+ if (entry.getKey().getName().toLowerCase(Locale.ROOT).equals(listenThread)
+ && isAccepting(entry.getValue()) == accepting)
+ {
+ return;
+ }
+ }
+ // 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"));
+ }
+
+ 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.
+ * <p>
+ * 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) 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, CHANGELOG_WRITER_RS_ID, 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 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();
+ }
+ 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<String> leftBehind;
+ while (!(leftBehind = domainRegistrationsOf(rsServerId)).isEmpty() && System.currentTimeMillis() < deadline)
+ {
+ // 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);
+ }
+
+ /** The threads a {@link ReplicationServerDomain} starts, and which its shutdown stops. */
+ private static final Collection<String> 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<String> domainRegistrationsOf(int rsServerId)
+ {
+ final String replicationServer = "replication server rs(" + rsServerId + ")";
+ final List<String> registrations = new ArrayList<>();
+ for (Thread thread : Thread.getAllStackTraces().keySet())
+ {
+ final String name = thread.getName().toLowerCase(Locale.ROOT);
+ 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<String> candidates)
+ {
+ for (String candidate : candidates)
+ {
+ if (name.contains(candidate))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Waits for the head log file of a replica changelog to appear under the provided changelog.
+ * <p>
+ * 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() + 60000;
+ while (findReplicaLogFile(dbDirectory) == null && System.currentTimeMillis() < deadline)
+ {
+ Thread.sleep(10);
+ }
+ assertNotNull(findReplicaLogFile(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");
+ }
+
+ /**
+ * Returns the head log file of a replica changelog, i.e. the one under a domain directory.
+ * <p>
+ * 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. */
+ 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<String> changelogVirtualAttributeNames()
{
--
Gitblit v1.10.0