From e82edf6c4244607b9fb8e38b5adb304a42bb7d33 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 3 Sep 2026 13:52:41 +0300 Subject: [PATCH 1/2] [#900] Time the ReplicaOfflineMsg grace period per replica, and spend it where the message can still be forwarded DSRSShutdownSync latched a single static timestamp on the first ReplicaOfflineMsg a process ever sent. A domain sends that message on every disableService() - an online import, a restore, a backend being disabled, a fractional or assured configuration change - so after the first one canShutdown() returned true at once for the rest of the process lifetime, and the shutdown the class exists for got no grace period at all. The other RSs were then never told the replica went offline, and their medium consistency point stayed pinned to its last CSN, so their change number index and external changelog stopped advancing for that domain. Keep the time of the message per domain and per replica. The collocated RS relays the message of every replica connected to it, so the forward of another replica's message says nothing about this one. The wait also had to move. ServerHandler.shutdown() deactivates the consumer, clears the message queue and closes the session before it joins the writer, so a writer looping on canShutdown() could no longer forward anything - and a ReplicaOfflineMsg is never written to the changelog, so the late queue cannot bring it back either. ReplicationServerDomain.shutdown() now waits before it stops its handlers, and only when an RS is connected to forward the message to; the loop in ServerWriter, which could only ever spin, is gone. --- .../plugin/LDAPReplicationDomain.java | 2 +- .../server/ReplicationServerDomain.java | 16 + .../replication/server/ServerWriter.java | 14 +- .../replication/service/DSRSShutdownSync.java | 155 +++++++- .../ReplicationServerShutdownSyncTest.java | 349 ++++++++++++++++++ .../service/DSRSShutdownSyncTest.java | 147 ++++++++ 6 files changed, 665 insertions(+), 18 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 2653da06fc..6d4229731e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -1976,7 +1976,7 @@ void doPreOperation(PreOperationAddOperation addOperation) public void publishReplicaOfflineMsg() { pendingChanges.putReplicaOfflineMsg(); - dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN()); + dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), getServerId()); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java index d8d24cef1c..cdd1a6b4ec 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java @@ -1666,6 +1666,22 @@ public void shutdown() { DirectoryServer.deregisterMonitorProvider(this); + /* + * Let a ReplicaOfflineMsg sent by a collocated DS be forwarded to the other RSs before the + * server handlers are stopped: stopping them deactivates their consumer, clears their message + * queue and closes their session, after which the message can no longer be sent - see + * OPENDJ-1453. With no RS connected there is nobody to forward the message to, and waiting + * would only delay the shutdown by the whole grace period. + *

+ * This waits before the assured timer is cancelled below, so that an assured update still + * waiting for acks keeps timing out during the wait instead of holding its sender until the + * sessions are closed. + */ + if (!connectedRSs.isEmpty()) + { + localReplicationServer.getDSRSShutdownSync().awaitReplicaOfflineMsgForwarded(baseDN); + } + // Terminate the assured timer assuredTimeoutTimer.cancel(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java index 2b7f685d02..5e2175b878 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java @@ -13,6 +13,7 @@ * * Copyright 2006-2009 Sun Microsystems, Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server; @@ -87,9 +88,15 @@ public void run() LocalizableMessage errMessage = null; try { + /* + * Looping here to wait for a pending ReplicaOfflineMsg would achieve nothing: this writer + * only stops once its handler has been shut down, which deactivates the consumer, clears + * the message queue and closes the session. The shutdown of the domain waits for the + * message to be forwarded before it stops the handlers - see + * ReplicationServerDomain.shutdown() and OPENDJ-1453. + */ boolean shutdown = false; - while (!shutdown - || !dsrsShutdownSync.canShutdown(replicationServerDomain.getBaseDN())) + while (!shutdown) { final UpdateMsg updateMsg = this.handler.take(); if (updateMsg == null) @@ -105,7 +112,8 @@ else if (!isUpdateMsgFiltered(updateMsg)) session.publish(updateMsg); if (updateMsg instanceof ReplicaOfflineMsg) { - dsrsShutdownSync.replicaOfflineMsgForwarded(replicationServerDomain.getBaseDN()); + dsrsShutdownSync.replicaOfflineMsgForwarded( + replicationServerDomain.getBaseDN(), updateMsg.getCSN().getServerId()); } } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java index 1b04bd0f52..2daa7a1460 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java @@ -12,11 +12,12 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.service; -import java.util.concurrent.ConcurrentSkipListSet; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import org.forgerock.opendj.ldap.DN; @@ -27,24 +28,80 @@ * More specifically, it ensures a ReplicaOfflineMsg sent by the DS is * relayed/forwarded by the collocated RS to the other RSs in the topology * before the whole process shuts down. + *

+ * The state is kept per domain and per instance: the collocated DS and RS + * sides coordinate through the single instance MultimasterReplication hands + * to both of them. * * @since OPENDJ-1453 */ public class DSRSShutdownSync { - private static final ConcurrentSkipListSet replicaOfflineMsgs = new ConcurrentSkipListSet<>(); - private static AtomicLong stopInstanceTimestamp = new AtomicLong(); + /** + * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated + * RS, in milliseconds, counted from the moment the message was sent. + */ + public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000; + + private final long gracePeriod; + + /** + * Time at which a ReplicaOfflineMsg was sent, per domain and per replica of + * that domain, for the messages which have not been forwarded yet. + *

+ * The time is kept per domain because a domain sends this message whenever + * its replication service is disabled - an online import, a restore, a + * configuration change - and not only when the process shuts down. A single + * time for the whole process would be the time of the first such message and + * would leave no grace period at all to the shutdown this class exists for. + *

+ * It is kept per replica because the collocated RS relays the message of + * every replica connected to it, and the forward of another replica's + * message says nothing about this one. + */ + private final ConcurrentMap> replicaOfflineMsgs = + new ConcurrentHashMap<>(); + /** Monitor notified whenever a ReplicaOfflineMsg has been forwarded. */ + private final Object forwardedMonitor = new Object(); + + /** Creates a synchronization object using the default grace period. */ + public DSRSShutdownSync() + { + this(REPLICA_OFFLINE_GRACE_PERIOD); + } + + /** + * Creates a synchronization object using the provided grace period. + * + * @param gracePeriod + * how long a ReplicaOfflineMsg may hold back the shutdown, in milliseconds + */ + DSRSShutdownSync(long gracePeriod) + { + this.gracePeriod = gracePeriod; + } /** * Message has been sent. * * @param baseDN * the domain for which the message has been sent + * @param serverId + * the replica which announced itself offline */ - public void replicaOfflineMsgSent(DN baseDN) + public void replicaOfflineMsgSent(DN baseDN, int serverId) { - stopInstanceTimestamp.compareAndSet(0, System.currentTimeMillis()); - replicaOfflineMsgs.add(baseDN); + ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs == null) + { + msgs = new ConcurrentHashMap<>(); + final ConcurrentMap existing = replicaOfflineMsgs.putIfAbsent(baseDN, msgs); + if (existing != null) + { + msgs = existing; + } + } + msgs.put(serverId, System.currentTimeMillis()); } /** @@ -52,23 +109,93 @@ public void replicaOfflineMsgSent(DN baseDN) * * @param baseDN * the domain for which the message has been sent + * @param serverId + * the replica the forwarded message belongs to */ - public void replicaOfflineMsgForwarded(DN baseDN) + public void replicaOfflineMsgForwarded(DN baseDN, int serverId) { - replicaOfflineMsgs.remove(baseDN); + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs != null) + { + msgs.remove(serverId); + } + synchronized (forwardedMonitor) + { + forwardedMonitor.notifyAll(); + } } /** - * Whether a ReplicationServer ServerReader or ServerWriter can proceed with - * shutdown. + * Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg + * has been forwarded or its grace period has expired. * * @param baseDN - * the baseDN of the ServerReader or ServerWriter . + * the baseDN of the domain being shut down * @return true if the caller can shutdown, false otherwise */ public boolean canShutdown(DN baseDN) { - return !replicaOfflineMsgs.contains(baseDN) - || System.currentTimeMillis() - stopInstanceTimestamp.get() > 5000; + return remainingGracePeriod(baseDN) <= 0; + } + + /** + * Waits for the ReplicaOfflineMsg of the provided domain to be forwarded, or for its grace + * period to expire. + *

+ * This must be called before the server handlers of the domain are stopped: stopping them + * deactivates their consumer, clears their message queue and closes their session, after which + * the message can no longer be forwarded. + * + * @param baseDN + * the baseDN of the domain whose message must be forwarded + */ + public void awaitReplicaOfflineMsgForwarded(DN baseDN) + { + // Bound the wait even if the domain keeps announcing itself offline while we are waiting. + final long deadline = System.currentTimeMillis() + gracePeriod; + synchronized (forwardedMonitor) + { + while (!canShutdown(baseDN)) + { + final long timeout = Math.min(remainingGracePeriod(baseDN), deadline - System.currentTimeMillis()); + if (timeout <= 0) + { + return; + } + try + { + forwardedMonitor.wait(timeout); + } + catch (InterruptedException e) + { + /* + * Give up waiting. The interrupt is deliberately not restored: what follows this call is + * the rest of the shutdown - joining the reader and writer thread of every handler, then + * closing the changelog DB - and an interrupt flag would make all of it give up too. + */ + return; + } + } + } + } + + /** + * Returns the time left to forward the ReplicaOfflineMsg of the replica of this domain which + * has the longest to wait, zero or less if no message of this domain is pending. + */ + private long remainingGracePeriod(DN baseDN) + { + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs == null) + { + return 0; + } + final long now = System.currentTimeMillis(); + long remaining = 0; + for (Long msgSentTime : msgs.values()) + { + remaining = Math.max(remaining, msgSentTime + gracePeriod - now); + } + return remaining; } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java new file mode 100644 index 0000000000..bc71a1b5f1 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java @@ -0,0 +1,349 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.server; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING; + +import java.net.ServerSocket; +import java.net.Socket; +import java.util.TreeSet; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.TestCaseUtils; +import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.protocol.ReplSessionSecurity; +import org.opends.server.replication.protocol.Session; +import org.opends.server.replication.service.DSRSShutdownSync; +import org.opends.server.replication.service.ReplicationBroker; +import org.opends.server.util.StaticUtils; +import org.testng.annotations.Test; + +/** + * The shutdown of a replication server must let a ReplicaOfflineMsg sent by a collocated + * directory server be forwarded to the other replication servers of the topology before the + * server handlers are stopped - stopping them deactivates their consumer, clears their message + * queue and closes their session, after which the message can no longer be sent. + *

+ * The tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory + * server: the contract they pin is when the shutdown of the replication server waits, and how + * long, without depending on the timing of a real session. + */ +@SuppressWarnings("javadoc") +public class ReplicationServerShutdownSyncTest extends ReplicationTestCase +{ + private static final int SOCKET_TIMEOUT_MS = 30000; + private static final int REMOTE_RS_ID = 92; + private static final int REMOTE_DS_ID = 93; + /** The collocated replica whose ReplicaOfflineMsg the shutdown waits for. */ + private static final int LOCAL_DS_ID = 94; + /** Time given to the forwarding thread before it releases the shutdown. */ + private static final long FORWARD_DELAY = 500; + + @Test + public void shutdownWaitsForTheReplicaOfflineMsgToBeForwarded() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncWaitDb", 8221); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + + final long startTime = System.currentTimeMillis(); + shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + replicationServer.shutdown(); + final long elapsed = System.currentTimeMillis() - startTime; + + assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + removeQuietly(replicationServer); + } + } + + @Test + public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final Thread forwarder = newForwarderThread(shutdownSync, baseDN); + ReplicationServer replicationServer = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncForwardDb", 8222); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + + final long startTime = System.currentTimeMillis(); + forwarder.start(); + replicationServer.shutdown(); + final long elapsed = System.currentTimeMillis() - startTime; + + assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY) + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + forwarder.join(); + removeQuietly(replicationServer); + } + } + + /** + * With no other replication server connected there is nobody to forward the message to, so + * waiting would only delay the shutdown of a standalone server by the whole grace period. + */ + @Test + public void shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try + { + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncAloneDb", 8223); + replicationServer.getReplicationServerDomain(baseDN, true); + + final long startTime = System.currentTimeMillis(); + shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + replicationServer.shutdown(); + final long elapsed = System.currentTimeMillis() - startTime; + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + removeQuietly(replicationServer); + } + } + + /** + * The writer serving a directory server must not hold back the shutdown either: it used to loop + * on the pending message until the grace period expired, although its handler had already been + * shut down and a ReplicaOfflineMsg is never sent to a directory server anyway. + */ + @Test + public void shutdownIsNotDelayedByTheWriterServingADirectoryServer() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + try + { + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort); + broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + + final long startTime = System.currentTimeMillis(); + shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + replicationServer.shutdown(); + final long elapsed = System.currentTimeMillis() - startTime; + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + stop(broker); + removeQuietly(replicationServer); + } + } + + @Test + public void shutdownIsNotDelayedWhenNoReplicaOfflineMsgIsPending() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncNoMsgDb", 8224); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + + final long startTime = System.currentTimeMillis(); + replicationServer.shutdown(); + final long elapsed = System.currentTimeMillis() - startTime; + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + removeQuietly(replicationServer); + } + } + + private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName, + int serverId) throws Exception + { + return newReplicationServer(shutdownSync, dbDirName, serverId, TestCaseUtils.findFreePort()); + } + + private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName, + int serverId, int replicationPort) throws Exception + { + return new ReplicationServer(new ReplServerFakeConfiguration( + replicationPort, dbDirName, 0, serverId, 0, 100, new TreeSet()), shutdownSync); + } + + /** Registers a peer replication server on the domain, exactly as the handshake does. */ + private void registerConnectedReplicationServer(ReplicationServer replicationServer, DN baseDN, + Session session) throws Exception + { + final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true); + final ReplicationServerHandler rsHandler = + new ReplicationServerHandler(session, 100, replicationServer, 100); + rsHandler.serverId = REMOTE_RS_ID; + rsHandler.serverURL = "127.0.0.1:1636"; + rsHandler.setBaseDNAndDomain(baseDN, false); + try + { + domain.lock(); + domain.register(rsHandler); + } + finally + { + domain.release(); + } + } + + private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, final DN baseDN) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + try + { + Thread.sleep(FORWARD_DELAY); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + return; + } + shutdownSync.replicaOfflineMsgForwarded(baseDN, LOCAL_DS_ID); + } + }); + } + + /** Teardown must never mask the primary assertion failure. */ + private void removeQuietly(ReplicationServer replicationServer) + { + try + { + remove(replicationServer); + } + catch (Exception ignored) + { + } + } + + /** + * Establishes a connected session pair over the given listen socket, as a remote server + * connecting to the RS would. The TLS negotiation performed by the session factories needs both + * ends handshaking at the same time, so the client end runs on its own thread. + * + * @return the two sessions: the remote (client) end first, then the local (server) end to hand + * to the handler under test + */ + private Session[] connectSessionPair(ServerSocket listenSocket, final ReplSessionSecurity security) + throws Exception + { + final Socket clientSocket = new Socket("127.0.0.1", listenSocket.getLocalPort()); + clientSocket.setTcpNoDelay(true); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + Future clientEnd = null; + boolean connected = false; + try + { + clientEnd = executor.submit(new Callable() + { + @Override + public Session call() throws Exception + { + return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS); + } + }); + + final Socket serverSocket = listenSocket.accept(); + serverSocket.setTcpNoDelay(true); + final Session serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS); + assertThat(serverEnd).as("could not create a session for the handler under test").isNotNull(); + + final Session[] sessionPair = + new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd }; + connected = true; + return sessionPair; + } + finally + { + if (!connected) + { + // Nobody owns the client end yet: close whatever it managed to create. + closeClientEndQuietly(clientEnd, clientSocket); + } + executor.shutdown(); + } + } + + private void closeClientEndQuietly(Future clientEnd, Socket clientSocket) + { + if (clientEnd != null) + { + try + { + final Session session = clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + if (session != null) + { + session.close(); + } + } + catch (Exception ignored) + { + clientEnd.cancel(true); + } + } + StaticUtils.close(clientSocket); + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java new file mode 100644 index 0000000000..9e3089af16 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java @@ -0,0 +1,147 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.service; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.DirectoryServerTestCase; +import org.opends.server.TestCaseUtils; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** Test the {@link DSRSShutdownSync} class. */ +@SuppressWarnings("javadoc") +public class DSRSShutdownSyncTest extends DirectoryServerTestCase +{ + /** Short grace period, so that the time dependent contracts are pinned in milliseconds. */ + private static final long GRACE_PERIOD = 200; + private static final int SERVER_ID = 1; + private static final int OTHER_SERVER_ID = 2; + + private static DN baseDN1; + private static DN baseDN2; + + @BeforeClass + public static void classSetup() throws Exception + { + TestCaseUtils.startServer(); + baseDN1 = DN.valueOf("dc=example,dc=com"); + baseDN2 = DN.valueOf("dc=world,dc=company"); + } + + @Test + public void canShutdownWhenNoReplicaOfflineMsgWasSent() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + @Test + public void cannotShutdownUntilTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + @Test + public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + @Test + public void canShutdownOnceTheGracePeriodExpired() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + Thread.sleep(GRACE_PERIOD + 50); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + /** + * A message sent earlier in the life of the process - an online import, a restore, a + * configuration change - must not consume the grace period of the message sent by the + * shutdown this class exists for. + */ + @Test + public void gracePeriodOfAShutdownIsNotSpentByAnEarlierMessage() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + // an import disables then re-enables the replication service + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); + Thread.sleep(GRACE_PERIOD + 50); + + // the shutdown of the process, much later + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + @Test + public void gracePeriodIsCountedPerDomain() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + Thread.sleep(GRACE_PERIOD + 50); + shutdownSync.replicaOfflineMsgSent(baseDN2, SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + assertThat(shutdownSync.canShutdown(baseDN2)).isFalse(); + } + + /** + * A replication server relays the ReplicaOfflineMsg of every replica connected to it, so the + * forward of another replica's message must not release the shutdown of this one. + */ + @Test + public void theForwardOfAnotherReplicasMessageDoesNotEndTheWait() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, OTHER_SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + @Test + public void aReplicaStillWaitingDoesNotHoldBackAnAlreadyForwardedOne() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, OTHER_SERVER_ID); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, OTHER_SERVER_ID); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } +} From 1b0a3c732c7854797b78a6297e74b597d4bdf581 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Sat, 5 Sep 2026 17:51:00 +0300 Subject: [PATCH 2/2] [#900] End the ReplicaOfflineMsg wait only on the forward it is waiting for Review of #919. * The wait was ended by any writer publishing a ReplicaOfflineMsg, a directory server's included. ReplicationServerDomain.put() never queues one for a directory server, yet a directory server which is catching up reads its updates from the changelog, where ReplicaCursor synthesizes one from the offline CSN of the replica. ServerWriter now releases the wait only for a peer replication server. * A replica announces itself offline on every disableService(), so the message forwarded now may be an older one, queued behind a backlog since an earlier import. The pending entry is keyed by the CSN of the announcement and a forward older than it no longer consumes its grace period: PendingChanges.putReplicaOfflineMsg() hands that CSN back for it. * All the domains of a shutdown wait together, before any of them is stopped, instead of one after the other: the grace period bounds the whole shutdown, and the wait of one domain no longer spends the grace period of the next. The wait runs on System.nanoTime(), so a backwards clock step cannot stretch it, and the path which aborts a failed initialization does not wait at all - nothing in an instance which never came up can forward a message. * Two comments recorded the opposite of what the code does and were corrected: the message can reach the writer serving a directory server, and the wait ends at the first forward, not once every peer has it (#917). The tests pin delivery and ordering, not only duration: a peer connected through the real handshake must have received the message before shutdown() returns, a message published to a directory server must not end the wait, a stale forward must not release a newer announcement, and one wait must cover every domain of the shutdown. Each of them was run against the behaviour it replaces and fails there. --- .../plugin/LDAPReplicationDomain.java | 4 +- .../replication/plugin/PendingChanges.java | 6 +- .../replication/server/ReplicationServer.java | 44 +- .../server/ReplicationServerDomain.java | 26 +- .../replication/server/ServerWriter.java | 20 +- .../replication/service/DSRSShutdownSync.java | 168 ++++-- .../ReplicationServerShutdownSyncTest.java | 514 ++++++++++++++++-- .../service/DSRSShutdownSyncTest.java | 194 ++++++- 8 files changed, 848 insertions(+), 128 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 6d4229731e..0cd82bb102 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -1975,8 +1975,8 @@ void doPreOperation(PreOperationAddOperation addOperation) @Override public void publishReplicaOfflineMsg() { - pendingChanges.putReplicaOfflineMsg(); - dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), getServerId()); + final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg(); + dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java index 5d2f6fdbe9..8ebba8cbd6 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java @@ -13,6 +13,7 @@ * * Copyright 2009 Sun Microsystems, Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; @@ -123,8 +124,10 @@ synchronized CSN putLocalOperation(PluginOperation operation) /** * Add a replica offline message to the pending list. + * + * @return the CSN of the message which was added */ - public synchronized void putReplicaOfflineMsg() + public synchronized CSN putReplicaOfflineMsg() { final CSN offlineCSN = csnGenerator.newCSN(); final PendingChange pendingChange = @@ -133,6 +136,7 @@ public synchronized void putReplicaOfflineMsg() pendingChanges.put(offlineCSN, pendingChange); pushCommittedChanges(); + return offlineCSN; } /** 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 4bea16769d..ef1ba8b22c 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 @@ -178,6 +178,11 @@ public class ReplicationServer /** * Creates a new Replication server using the provided configuration entry. + *

+ * The synchronization object this creates is its own, so the resulting server does not + * synchronize its shutdown with a collocated directory server. A server which has to must be + * built with {@link #ReplicationServer(ReplicationServerCfg, DSRSShutdownSync)}, passing the + * instance the directory server side records its ReplicaOfflineMsgs on. * * @param cfg The configuration of this replication server. * @throws ConfigException When Configuration is invalid. @@ -802,6 +807,9 @@ private void abortInitialization() // 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. + // Nothing in an instance which never finished coming up can forward a pending + // ReplicaOfflineMsg, so this path does not wait for one: it would only delay the failure + // which is being reported by a grace period which cannot pay off. for (ReplicationServerDomain domain : getReplicationServerDomains()) { try @@ -1181,7 +1189,19 @@ public void shutdown() listenThread.interrupt(); } - // shutdown all the replication domains + /* + * Let the ReplicaOfflineMsgs a collocated DS sent be forwarded while every handler is still + * up, and only then stop the domains: shutting a domain down deactivates the consumer of its + * handlers, clears their message queue and closes their session - see OPENDJ-1453. All the + * domains wait together and share one deadline, so the shutdown is bounded by one grace + * period and the wait of one domain does not spend the grace period of the next. + *

+ * This also runs before the assured timer of any domain is cancelled, so an assured update + * still waiting for acks keeps timing out during the wait instead of holding its sender + * until the sessions are closed. + */ + awaitReplicaOfflineMsgsForwarded(); + for (ReplicationServerDomain domain : getReplicationServerDomains()) { domain.shutdown(); @@ -1202,6 +1222,28 @@ public void shutdown() allInstances.remove(this); } + /** + * Waits for the ReplicaOfflineMsg of every domain which has a replication server to forward it + * to. With no such server connected there is nobody to forward the message to, and waiting + * would only delay the shutdown by the whole grace period. + */ + private void awaitReplicaOfflineMsgsForwarded() + { + final List domainsToWaitFor = new ArrayList<>(); + for (ReplicationServerDomain domain : getReplicationServerDomains()) + { + if (!domain.getConnectedRSs().isEmpty()) + { + domainsToWaitFor.add(domain.getBaseDN()); + } + } + if (!domainsToWaitFor.isEmpty()) + { + dsrsShutdownSync.awaitReplicaOfflineMsgsForwarded( + domainsToWaitFor, dsrsShutdownSync.newShutdownDeadline()); + } + } + /** * Retrieves the time after which changes must be deleted from the * persistent storage (in milliseconds). diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java index cdd1a6b4ec..1674f4c5c8 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java @@ -1661,27 +1661,19 @@ private MonitorMsg createLocalTopologyMonitorMsg(int sender, int destination) return monitorMsg; } - /** Shutdown this ReplicationServerDomain. */ + /** + * Shutdown this ReplicationServerDomain. + *

+ * A ReplicaOfflineMsg which a collocated DS sent and which is still to be forwarded must be + * waited for before this runs: stopping the server handlers deactivates their consumer, clears + * their message queue and closes their session, after which the message can no longer be sent + * - see OPENDJ-1453. ReplicationServer.shutdown() waits for the messages of all of its domains + * before it stops any of them. + */ public void shutdown() { DirectoryServer.deregisterMonitorProvider(this); - /* - * Let a ReplicaOfflineMsg sent by a collocated DS be forwarded to the other RSs before the - * server handlers are stopped: stopping them deactivates their consumer, clears their message - * queue and closes their session, after which the message can no longer be sent - see - * OPENDJ-1453. With no RS connected there is nobody to forward the message to, and waiting - * would only delay the shutdown by the whole grace period. - *

- * This waits before the assured timer is cancelled below, so that an assured update still - * waiting for acks keeps timing out during the wait instead of holding its sender until the - * sessions are closed. - */ - if (!connectedRSs.isEmpty()) - { - localReplicationServer.getDSRSShutdownSync().awaitReplicaOfflineMsgForwarded(baseDN); - } - // Terminate the assured timer assuredTimeoutTimer.cancel(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java index 5e2175b878..aa9dc21805 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java @@ -95,8 +95,7 @@ public void run() * message to be forwarded before it stops the handlers - see * ReplicationServerDomain.shutdown() and OPENDJ-1453. */ - boolean shutdown = false; - while (!shutdown) + while (true) { final UpdateMsg updateMsg = this.handler.take(); if (updateMsg == null) @@ -104,16 +103,25 @@ public void run() // this connection is closing errMessage = LocalizableMessage.raw( "Connection closure: null update returned by domain."); - shutdown = true; + break; } - else if (!isUpdateMsgFiltered(updateMsg)) + if (!isUpdateMsgFiltered(updateMsg)) { // Publish the update to the remote server using a protocol version it supports session.publish(updateMsg); - if (updateMsg instanceof ReplicaOfflineMsg) + /* + * Only the forward to a peer RS ends the wait of the shutdown: what the grace period + * buys is the rest of the topology learning that the replica went offline. + * ReplicationServerDomain.put() never queues this message for a directory server - its + * isUpdateMsgFiltered() drops it there - but a directory server which is catching up + * reads its updates from the changelog, where ReplicaCursor synthesizes a + * ReplicaOfflineMsg from the offline CSN of the replica. Publishing that one says + * nothing about the peer RSs the shutdown is waiting for. + */ + if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer()) { dsrsShutdownSync.replicaOfflineMsgForwarded( - replicationServerDomain.getBaseDN(), updateMsg.getCSN().getServerId()); + replicationServerDomain.getBaseDN(), updateMsg.getCSN()); } } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java index 2daa7a1460..3d962b0e08 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java @@ -16,10 +16,15 @@ */ package org.opends.server.replication.service; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import java.util.Collection; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import org.forgerock.opendj.ldap.DN; +import org.opends.server.replication.common.CSN; /** * Class useful for the case where DS/RS instances are collocated inside the @@ -46,20 +51,20 @@ public class DSRSShutdownSync private final long gracePeriod; /** - * Time at which a ReplicaOfflineMsg was sent, per domain and per replica of - * that domain, for the messages which have not been forwarded yet. + * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per + * replica of that domain. *

- * The time is kept per domain because a domain sends this message whenever - * its replication service is disabled - an online import, a restore, a + * It is kept per domain because a domain sends this message whenever its + * replication service is disabled - an online import, a restore, a * configuration change - and not only when the process shuts down. A single - * time for the whole process would be the time of the first such message and + * entry for the whole process would be the one of the first such message and * would leave no grace period at all to the shutdown this class exists for. *

* It is kept per replica because the collocated RS relays the message of * every replica connected to it, and the forward of another replica's * message says nothing about this one. */ - private final ConcurrentMap> replicaOfflineMsgs = + private final ConcurrentMap> replicaOfflineMsgs = new ConcurrentHashMap<>(); /** Monitor notified whenever a ReplicaOfflineMsg has been forwarded. */ private final Object forwardedMonitor = new Object(); @@ -86,22 +91,15 @@ public DSRSShutdownSync() * * @param baseDN * the domain for which the message has been sent - * @param serverId - * the replica which announced itself offline + * @param offlineCSN + * the CSN of the message, which identifies both the replica which announced itself + * offline and the announcement being waited for */ - public void replicaOfflineMsgSent(DN baseDN, int serverId) + public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN) { - ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); - if (msgs == null) - { - msgs = new ConcurrentHashMap<>(); - final ConcurrentMap existing = replicaOfflineMsgs.putIfAbsent(baseDN, msgs); - if (existing != null) - { - msgs = existing; - } - } - msgs.put(serverId, System.currentTimeMillis()); + replicaOfflineMsgs + .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap()) + .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime())); } /** @@ -109,15 +107,27 @@ public void replicaOfflineMsgSent(DN baseDN, int serverId) * * @param baseDN * the domain for which the message has been sent - * @param serverId - * the replica the forwarded message belongs to + * @param forwardedCSN + * the CSN of the forwarded message */ - public void replicaOfflineMsgForwarded(DN baseDN, int serverId) + public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN) { - final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); if (msgs != null) { - msgs.remove(serverId); + final int serverId = forwardedCSN.getServerId(); + final PendingOfflineMsg pending = msgs.get(serverId); + /* + * A replica announces itself offline on every disableService(), so the message which is + * forwarded now may be an older one - queued behind a backlog since an earlier import, or + * synthesized from the offline CSN of the changelog for a server which is catching up. + * Such a forward says nothing about the announcement the shutdown is waiting for, and must + * not consume its grace period. + */ + if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN)) + { + msgs.remove(serverId, pending); + } } synchronized (forwardedMonitor) { @@ -128,10 +138,14 @@ public void replicaOfflineMsgForwarded(DN baseDN, int serverId) /** * Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg * has been forwarded or its grace period has expired. + *

+ * The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)} + * rather than polling this; it is the same state, observable without waiting for it. * * @param baseDN * the baseDN of the domain being shut down - * @return true if the caller can shutdown, false otherwise + * @return true if the shutdown of this domain need not wait any longer, i.e. its message was + * forwarded or its grace period has expired, false otherwise */ public boolean canShutdown(DN baseDN) { @@ -139,25 +153,52 @@ public boolean canShutdown(DN baseDN) } /** - * Waits for the ReplicaOfflineMsg of the provided domain to be forwarded, or for its grace - * period to expire. + * Returns the time by which every wait of one shutdown must be over. *

- * This must be called before the server handlers of the domain are stopped: stopping them + * A process shuts its domains down one after the other and each of them may have a message + * pending, so a deadline computed once and shared by all of them keeps the whole shutdown + * bounded by one grace period instead of one per domain. + * + * @return the point in time, on the {@link System#nanoTime()} clock, by which the waits must + * be over + */ + public long newShutdownDeadline() + { + return System.nanoTime() + MILLISECONDS.toNanos(gracePeriod); + } + + /** + * Waits for the ReplicaOfflineMsg of every provided domain to be forwarded, or for their grace + * periods or the provided deadline to expire. + *

+ * This must be called before the server handlers of those domains are stopped: stopping them * deactivates their consumer, clears their message queue and closes their session, after which * the message can no longer be forwarded. + *

+ * All the domains of one shutdown wait together rather than one after the other, so that the + * shutdown is bounded by one grace period without the wait of one domain spending the grace + * period of the next. * - * @param baseDN - * the baseDN of the domain whose message must be forwarded + * @param baseDNs + * the baseDNs of the domains whose messages must be forwarded + * @param deadline + * the point in time, on the {@link System#nanoTime()} clock, by which this wait must + * be over whatever the domains announce in the meantime - see + * {@link #newShutdownDeadline()}. A deadline which is not in the future returns + * without waiting at all, for a caller which has nothing to wait for. */ - public void awaitReplicaOfflineMsgForwarded(DN baseDN) + public void awaitReplicaOfflineMsgsForwarded(Collection baseDNs, long deadline) { - // Bound the wait even if the domain keeps announcing itself offline while we are waiting. - final long deadline = System.currentTimeMillis() + gracePeriod; + if (deadline - System.nanoTime() <= 0) + { + return; + } synchronized (forwardedMonitor) { - while (!canShutdown(baseDN)) + while (true) { - final long timeout = Math.min(remainingGracePeriod(baseDN), deadline - System.currentTimeMillis()); + final long timeout = Math.min(remainingGracePeriod(baseDNs), + NANOSECONDS.toMillis(deadline - System.nanoTime())); if (timeout <= 0) { return; @@ -180,22 +221,65 @@ public void awaitReplicaOfflineMsgForwarded(DN baseDN) } /** - * Returns the time left to forward the ReplicaOfflineMsg of the replica of this domain which - * has the longest to wait, zero or less if no message of this domain is pending. + * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of + * the provided domains which has the longest to wait, zero or less if none of them has a + * message pending. + */ + private long remainingGracePeriod(Collection baseDNs) + { + long remaining = 0; + for (DN baseDN : baseDNs) + { + remaining = Math.max(remaining, remainingGracePeriod(baseDN)); + } + return remaining; + } + + /** + * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of + * this domain which has the longest to wait, zero or less if no message of this domain is + * pending. */ private long remainingGracePeriod(DN baseDN) { - final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); if (msgs == null) { return 0; } - final long now = System.currentTimeMillis(); + final long now = System.nanoTime(); long remaining = 0; - for (Long msgSentTime : msgs.values()) + for (PendingOfflineMsg pending : msgs.values()) { - remaining = Math.max(remaining, msgSentTime + gracePeriod - now); + remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime)); } return remaining; } + + /** + * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet. + *

+ * This deliberately does not override {@code equals}: the two-argument + * {@link ConcurrentMap#remove(Object, Object)} of the forward guard must match the very + * announcement it read, not another one which happens to carry the same values. + */ + private static final class PendingOfflineMsg + { + /** The CSN of the message, so that the forward of an older one is not taken for this one. */ + private final CSN csn; + /** When the message was announced, on the {@link System#nanoTime()} clock. */ + private final long sentTime; + + private PendingOfflineMsg(CSN csn, long sentTime) + { + this.csn = csn; + this.sentTime = sentTime; + } + + @Override + public String toString() + { + return "PendingOfflineMsg(" + csn + ")"; + } + } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java index bc71a1b5f1..18c7002f85 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java @@ -17,7 +17,9 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING; +import static org.opends.server.util.CollectionUtils.newArrayList; +import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; import java.util.TreeSet; @@ -26,15 +28,25 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.forgerock.opendj.ldap.DN; import org.opends.server.TestCaseUtils; import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.common.RSInfo; +import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.ReplServerStartMsg; import org.opends.server.replication.protocol.ReplSessionSecurity; +import org.opends.server.replication.protocol.ReplicaOfflineMsg; +import org.opends.server.replication.protocol.ReplicationMsg; import org.opends.server.replication.protocol.Session; +import org.opends.server.replication.protocol.TopologyMsg; import org.opends.server.replication.service.DSRSShutdownSync; import org.opends.server.replication.service.ReplicationBroker; import org.opends.server.util.StaticUtils; +import org.opends.server.util.TestTimer; import org.testng.annotations.Test; /** @@ -43,20 +55,26 @@ * server handlers are stopped - stopping them deactivates their consumer, clears their message * queue and closes their session, after which the message can no longer be sent. *

- * The tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory + * Most tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory * server: the contract they pin is when the shutdown of the replication server waits, and how - * long, without depending on the timing of a real session. + * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} pins the outcome + * those waits exist for, on a peer connected through the real handshake. */ @SuppressWarnings("javadoc") public class ReplicationServerShutdownSyncTest extends ReplicationTestCase { private static final int SOCKET_TIMEOUT_MS = 30000; + /** A session end nobody owns is discarded, so its cleanup waits far less than a live one. */ + private static final int DISCARDED_SESSION_TIMEOUT_MS = 2000; private static final int REMOTE_RS_ID = 92; private static final int REMOTE_DS_ID = 93; /** The collocated replica whose ReplicaOfflineMsg the shutdown waits for. */ private static final int LOCAL_DS_ID = 94; /** Time given to the forwarding thread before it releases the shutdown. */ private static final long FORWARD_DELAY = 500; + /** How often the domains of {@link #theGracePeriodIsSharedByAllTheDomainsOfOneShutdown()} + * announce themselves offline again while the shutdown is waiting for them. */ + private static final long REANNOUNCE_INTERVAL = 200; @Test public void shutdownWaitsForTheReplicaOfflineMsgToBeForwarded() throws Exception @@ -74,10 +92,10 @@ public void shutdownWaitsForTheReplicaOfflineMsgToBeForwarded() throws Exception { registerConnectedReplicationServer(replicationServer, baseDN, session); - final long startTime = System.currentTimeMillis(); - shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); replicationServer.shutdown(); - final long elapsed = System.currentTimeMillis() - startTime; + final long elapsed = elapsedMillis(startTime); assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); } @@ -93,8 +111,8 @@ public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exce { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); - final Thread forwarder = newForwarderThread(shutdownSync, baseDN); ReplicationServer replicationServer = null; + Thread forwarder = null; try (ServerSocket listen = TestCaseUtils.bindFreePort()) { listen.setSoTimeout(SOCKET_TIMEOUT_MS); @@ -104,12 +122,14 @@ public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exce Session session = sessionPair[1]) { registerConnectedReplicationServer(replicationServer, baseDN, session); - shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + final CSN offlineCSN = newOfflineCSN(); + forwarder = newForwarderThread(shutdownSync, baseDN, offlineCSN); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); - final long startTime = System.currentTimeMillis(); + final long startTime = System.nanoTime(); forwarder.start(); replicationServer.shutdown(); - final long elapsed = System.currentTimeMillis() - startTime; + final long elapsed = elapsedMillis(startTime); assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY) .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); @@ -117,7 +137,130 @@ public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exce } finally { - forwarder.join(); + joinQuietly(forwarder); + removeQuietly(replicationServer); + } + } + + /** + * The outcome the grace period exists for, end to end: a peer replication server connected + * through the real handshake has received the ReplicaOfflineMsg of the collocated replica by + * the time the shutdown returns. + *

+ * The waiting tests above measure durations only, so they stay green if the wait is moved + * after the handlers are stopped - which reintroduces OPENDJ-1453 and loses the message. This + * one fails in that case. + */ + @Test + public void thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + FakePeerReplicationServer peer = null; + Thread publisher = null; + try + { + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDeliveryDb", 8226, replicationPort); + broker = openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + peer = new FakePeerReplicationServer(replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID); + + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + waitForConnectedReplicationServer(domain); + final Future received = peer.receiveReplicaOfflineMsg(); + + /* + * The replica announces itself offline once the shutdown of the replication server is + * already waiting for the message, which is the ordering the grace period exists for. + */ + final CSN offlineCSN = newOfflineCSN(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + publisher = newPublisherThread(broker, offlineCSN); + + final long startTime = System.nanoTime(); + publisher.start(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertThat(forwarded) + .as("the peer replication server was never told that the replica went offline, its " + + "read ended with: %s", peer.readerFailure()) + .isNotNull(); + assertThat(forwarded.getCSN().getServerId()).isEqualTo(LOCAL_DS_ID); + assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY) + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + joinQuietly(publisher); + closeQuietly(peer); + stop(broker); + removeQuietly(replicationServer); + } + } + + /** + * Only a peer replication server learning about the offline replica ends the wait. + * ReplicationServerDomain.put() never queues a ReplicaOfflineMsg for a directory server, but + * the changelog cursor of a directory server which is catching up synthesizes one from the + * offline CSN of the replica, so the writer serving a directory server can publish it - and + * the peer replication servers would still know nothing. + */ + @Test + public void theForwardToADirectoryServerDoesNotEndTheWait() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDataServerForwardDb", 8227, replicationPort); + broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + // a peer replication server, so that the shutdown does wait for the message: what this + // test pins is that the directory server receiving it is not what ends that wait + registerConnectedReplicationServer(replicationServer, baseDN, session); + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + final DataServerHandler dsHandler = waitForConnectedDirectoryServer(domain); + + final CSN offlineCSN = newOfflineCSN(); + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + // the very message the shutdown waits for, so only the guard of the writer can save it + dsHandler.add(new ReplicaOfflineMsg(offlineCSN)); + + // the directory server did receive it, so its writer went through the forwarding code + assertThat(waitForSpecificMsg(broker, ReplicaOfflineMsg.class).getCSN().getServerId()) + .isEqualTo(LOCAL_DS_ID); + assertThat(elapsedMillis(startTime)) + .as("the fixture must deliver the message well inside the grace period, otherwise " + + "the wait asserted below cannot be told apart from a slow delivery") + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD / 2); + + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed) + .as("the message published to a directory server ended the wait of the shutdown") + .isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + stop(broker); removeQuietly(replicationServer); } } @@ -137,10 +280,10 @@ public void shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage replicationServer = newReplicationServer(shutdownSync, "shutdownSyncAloneDb", 8223); replicationServer.getReplicationServerDomain(baseDN, true); - final long startTime = System.currentTimeMillis(); - shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); replicationServer.shutdown(); - final long elapsed = System.currentTimeMillis() - startTime; + final long elapsed = elapsedMillis(startTime); assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); } @@ -151,9 +294,9 @@ public void shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage } /** - * The writer serving a directory server must not hold back the shutdown either: it used to loop - * on the pending message until the grace period expired, although its handler had already been - * shut down and a ReplicaOfflineMsg is never sent to a directory server anyway. + * The writer serving a directory server must not hold back the shutdown either: it used to + * loop on the pending message until the grace period expired, although its handler had already + * been shut down - which deactivates its consumer and leaves the loop nothing to take. */ @Test public void shutdownIsNotDelayedByTheWriterServingADirectoryServer() throws Exception @@ -169,10 +312,10 @@ public void shutdownIsNotDelayedByTheWriterServingADirectoryServer() throws Exce newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort); broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); - final long startTime = System.currentTimeMillis(); - shutdownSync.replicaOfflineMsgSent(baseDN, LOCAL_DS_ID); + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); replicationServer.shutdown(); - final long elapsed = System.currentTimeMillis() - startTime; + final long elapsed = elapsedMillis(startTime); assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); } @@ -199,9 +342,9 @@ public void shutdownIsNotDelayedWhenNoReplicaOfflineMsgIsPending() throws Except { registerConnectedReplicationServer(replicationServer, baseDN, session); - final long startTime = System.currentTimeMillis(); + final long startTime = System.nanoTime(); replicationServer.shutdown(); - final long elapsed = System.currentTimeMillis() - startTime; + final long elapsed = elapsedMillis(startTime); assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); } @@ -212,6 +355,67 @@ public void shutdownIsNotDelayedWhenNoReplicaOfflineMsgIsPending() throws Except } } + /** + * The domains of a replication server are shut down one after the other, so the grace period + * must bound the whole shutdown and not each of its domains: a process with several base DNs + * would otherwise pay it once per domain. + *

+ * Both domains keep announcing themselves offline while the shutdown is running, so neither + * wait can be ended by a forward and each of them runs to its bound - one grace period in + * total if it is shared, one per domain otherwise. + */ + @Test + public void theGracePeriodIsSharedByAllTheDomainsOfOneShutdown() throws Exception + { + final DN baseDN1 = DN.valueOf(TEST_ROOT_DN_STRING); + final DN baseDN2 = DN.valueOf("dc=world,dc=company"); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final AtomicBoolean stopped = new AtomicBoolean(); + ReplicationServer replicationServer = null; + Thread reAnnouncer = null; + try (ServerSocket listen1 = TestCaseUtils.bindFreePort(); + ServerSocket listen2 = TestCaseUtils.bindFreePort()) + { + listen1.setSoTimeout(SOCKET_TIMEOUT_MS); + listen2.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncSharedDeadlineDb", 8228); + final Session[] sessionPair1 = connectSessionPair(listen1, getReplSessionSecurity()); + final Session[] sessionPair2 = connectSessionPair(listen2, getReplSessionSecurity()); + try (Session remoteEnd1 = sessionPair1[0]; + Session session1 = sessionPair1[1]; + Session remoteEnd2 = sessionPair2[0]; + Session session2 = sessionPair2[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN1, session1); + registerConnectedReplicationServer(replicationServer, baseDN2, session2); + /* + * Announce both domains offline here rather than leaving it to the thread below: the + * wait of the shutdown must be armed whatever that thread has had time to run. + */ + final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0); + shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN()); + shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN()); + reAnnouncer = newReAnnouncerThread(shutdownSync, baseDN1, baseDN2, stopped); + reAnnouncer.start(); + + final long startTime = System.nanoTime(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + assertThat(elapsed) + .as("each domain waited its own grace period instead of sharing one deadline") + .isLessThan(2 * DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + stopped.set(true); + joinQuietly(reAnnouncer); + removeQuietly(replicationServer); + } + } + private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName, int serverId) throws Exception { @@ -225,9 +429,13 @@ private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, St replicationPort, dbDirName, 0, serverId, 0, 100, new TreeSet()), shutdownSync); } - /** Registers a peer replication server on the domain, exactly as the handshake does. */ - private void registerConnectedReplicationServer(ReplicationServer replicationServer, DN baseDN, - Session session) throws Exception + /** + * Registers a peer replication server on the domain as the handshake does, but without the + * protocol exchange: the handler this leaves behind has no writer, which is enough for the + * tests which only need a domain with a connected peer. + */ + private void registerConnectedReplicationServer( + ReplicationServer replicationServer, DN baseDN, Session session) throws Exception { final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true); final ReplicationServerHandler rsHandler = @@ -235,9 +443,9 @@ private void registerConnectedReplicationServer(ReplicationServer replicationSer rsHandler.serverId = REMOTE_RS_ID; rsHandler.serverURL = "127.0.0.1:1636"; rsHandler.setBaseDNAndDomain(baseDN, false); + domain.lock(); try { - domain.lock(); domain.register(rsHandler); } finally @@ -246,27 +454,124 @@ private void registerConnectedReplicationServer(ReplicationServer replicationSer } } - private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, final DN baseDN) + private void waitForConnectedReplicationServer(final ReplicationServerDomain domain) + throws Exception + { + newConnectionTimer().repeatUntilSuccess(new TestTimer.CallableVoid() + { + @Override + public void call() throws Exception + { + assertThat(domain.getConnectedRSs()) + .as("the peer replication server never connected").containsKey(REMOTE_RS_ID); + } + }); + } + + private DataServerHandler waitForConnectedDirectoryServer(final ReplicationServerDomain domain) + throws Exception + { + return newConnectionTimer().repeatUntilSuccess(new Callable() + { + @Override + public DataServerHandler call() throws Exception + { + final DataServerHandler dsHandler = domain.getConnectedDSs().get(REMOTE_DS_ID); + assertThat(dsHandler).as("the directory server never connected").isNotNull(); + return dsHandler; + } + }); + } + + private static TestTimer newConnectionTimer() + { + return new TestTimer.Builder() + .maxSleep(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS) + .sleepTimes(10, TimeUnit.MILLISECONDS) + .toTimer(); + } + + private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, final DN baseDN, + final CSN offlineCSN) { return new Thread(new Runnable() { @Override public void run() { - try + if (!sleepQuietly(FORWARD_DELAY)) { - Thread.sleep(FORWARD_DELAY); + return; } - catch (InterruptedException e) + shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN); + } + }); + } + + private Thread newPublisherThread(final ReplicationBroker broker, final CSN offlineCSN) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + if (!sleepQuietly(FORWARD_DELAY)) { - Thread.currentThread().interrupt(); return; } - shutdownSync.replicaOfflineMsgForwarded(baseDN, LOCAL_DS_ID); + broker.publish(new ReplicaOfflineMsg(offlineCSN)); + } + }); + } + + private Thread newReAnnouncerThread(final DSRSShutdownSync shutdownSync, final DN baseDN1, + final DN baseDN2, final AtomicBoolean stopped) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0); + while (!stopped.get()) + { + shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN()); + shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN()); + if (!sleepQuietly(REANNOUNCE_INTERVAL)) + { + return; + } + } } }); } + /** Milliseconds elapsed since a {@link System#nanoTime()} reading, the clock the waits use. */ + private static long elapsedMillis(long startTime) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); + } + + /** The CSN of a message the collocated replica announces, as PendingChanges generates it. */ + private static CSN newOfflineCSN() + { + return new CSNGenerator(LOCAL_DS_ID, 0).newCSN(); + } + + private static boolean sleepQuietly(long millis) + { + try + { + Thread.sleep(millis); + return true; + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + return false; + } + } + /** Teardown must never mask the primary assertion failure. */ private void removeQuietly(ReplicationServer replicationServer) { @@ -279,6 +584,29 @@ private void removeQuietly(ReplicationServer replicationServer) } } + private void joinQuietly(Thread thread) + { + if (thread != null) + { + try + { + thread.join(SOCKET_TIMEOUT_MS); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + } + + private void closeQuietly(FakePeerReplicationServer peer) + { + if (peer != null) + { + peer.close(); + } + } + /** * Establishes a connected session pair over the given listen socket, as a remote server * connecting to the RS would. The TLS negotiation performed by the session factories needs both @@ -294,6 +622,8 @@ private Session[] connectSessionPair(ServerSocket listenSocket, final ReplSessio clientSocket.setTcpNoDelay(true); final ExecutorService executor = Executors.newSingleThreadExecutor(); Future clientEnd = null; + Socket serverSocket = null; + Session serverEnd = null; boolean connected = false; try { @@ -306,9 +636,9 @@ public Session call() throws Exception } }); - final Socket serverSocket = listenSocket.accept(); + serverSocket = listenSocket.accept(); serverSocket.setTcpNoDelay(true); - final Session serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS); + serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS); assertThat(serverEnd).as("could not create a session for the handler under test").isNotNull(); final Session[] sessionPair = @@ -320,20 +650,33 @@ public Session call() throws Exception { if (!connected) { - // Nobody owns the client end yet: close whatever it managed to create. + // Nobody owns either end yet: close whatever they managed to create. closeClientEndQuietly(clientEnd, clientSocket); + closeServerEndQuietly(serverEnd, serverSocket); } executor.shutdown(); } } + private void closeServerEndQuietly(Session serverEnd, Socket serverSocket) + { + if (serverEnd != null) + { + serverEnd.close(); + } + else + { + StaticUtils.close(serverSocket); + } + } + private void closeClientEndQuietly(Future clientEnd, Socket clientSocket) { if (clientEnd != null) { try { - final Session session = clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + final Session session = clientEnd.get(DISCARDED_SESSION_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (session != null) { session.close(); @@ -346,4 +689,105 @@ private void closeClientEndQuietly(Future clientEnd, Socket clientSocke } StaticUtils.close(clientSocket); } + + /** + * A peer replication server which connects to the replication server under test and completes + * the handshake, so that the handler it leaves behind on the domain has a real writer and can + * actually forward what the domain pushes to it. + */ + private static final class FakePeerReplicationServer + { + private final Session session; + private final ExecutorService reader = Executors.newSingleThreadExecutor(); + /** Why the peer stopped reading, so that a missing message can be told from a failed one. */ + private volatile Exception readerFailure; + + FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId) + throws Exception + { + final Socket socket = new Socket(); + Session newSession = null; + boolean handshaken = false; + try + { + socket.setTcpNoDelay(true); + socket.connect(new InetSocketAddress("127.0.0.1", replicationPort), SOCKET_TIMEOUT_MS); + newSession = getReplSessionSecurity().createClientSession(socket, SOCKET_TIMEOUT_MS); + + final String serverURL = "127.0.0.1:" + socket.getLocalPort(); + final byte groupId = (byte) 1; + newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, 100, + new ServerState(), generationId, false, groupId, 5000)); + final ReplServerStartMsg inStartMsg = + waitForSpecificMsg(newSession, ReplServerStartMsg.class); + if (!inStartMsg.getSSLEncryption()) + { + newSession.stopEncryption(); + } + newSession.publish(new TopologyMsg(null, + newArrayList(new RSInfo(serverId, serverURL, generationId, groupId, 1)))); + waitForSpecificMsg(newSession, TopologyMsg.class); + handshaken = true; + } + finally + { + if (!handshaken) + { + // The caller has no handle on this peer yet, so nothing else would close it. + reader.shutdownNow(); + if (newSession != null) + { + newSession.close(); + } + else + { + StaticUtils.close(socket); + } + } + } + session = newSession; + } + + /** Returns the first ReplicaOfflineMsg this peer receives, or null if its session ends first. */ + Future receiveReplicaOfflineMsg() + { + return reader.submit(new Callable() + { + @Override + public ReplicaOfflineMsg call() + { + try + { + while (true) + { + final ReplicationMsg msg = session.receive(); + if (msg instanceof ReplicaOfflineMsg) + { + return (ReplicaOfflineMsg) msg; + } + } + } + catch (Exception e) + { + // The session is closed when the replication server completes its shutdown: whatever + // has not arrived by then never will. + readerFailure = e; + return null; + } + } + }); + } + + /** Returns what ended the read of this peer, null if nothing did. */ + Exception readerFailure() + { + return readerFailure; + } + + void close() + { + reader.shutdownNow(); + session.close(); + } + } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java index 9e3089af16..81408a0a87 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java @@ -15,11 +15,14 @@ */ package org.opends.server.replication.service; +import static java.util.Arrays.asList; import static org.assertj.core.api.Assertions.assertThat; +import java.util.concurrent.TimeUnit; + import org.forgerock.opendj.ldap.DN; import org.opends.server.DirectoryServerTestCase; -import org.opends.server.TestCaseUtils; +import org.opends.server.replication.common.CSN; import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @@ -27,8 +30,15 @@ @SuppressWarnings("javadoc") public class DSRSShutdownSyncTest extends DirectoryServerTestCase { - /** Short grace period, so that the time dependent contracts are pinned in milliseconds. */ - private static final long GRACE_PERIOD = 200; + /** Short grace period, for the contracts a test has to wait out. */ + private static final long GRACE_PERIOD = 500; + /** + * Grace period for the contracts which only read the state: long enough that no scheduling + * pause between announcing a message and reading the state can expire it. + */ + private static final long LONG_GRACE_PERIOD = 60000; + /** Time given to the forwarding thread before it forwards the message of one domain. */ + private static final long FORWARD_DELAY = 200; private static final int SERVER_ID = 1; private static final int OTHER_SERVER_ID = 2; @@ -38,7 +48,6 @@ public class DSRSShutdownSyncTest extends DirectoryServerTestCase @BeforeClass public static void classSetup() throws Exception { - TestCaseUtils.startServer(); baseDN1 = DN.valueOf("dc=example,dc=com"); baseDN2 = DN.valueOf("dc=world,dc=company"); } @@ -54,9 +63,9 @@ public void canShutdownWhenNoReplicaOfflineMsgWasSent() throws Exception @Test public void cannotShutdownUntilTheReplicaOfflineMsgIsForwarded() throws Exception { - final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); } @@ -64,10 +73,11 @@ public void cannotShutdownUntilTheReplicaOfflineMsgIsForwarded() throws Exceptio @Test public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception { - final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN offlineCSN = newCSN(SERVER_ID); - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); - shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN); assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); } @@ -77,7 +87,7 @@ public void canShutdownOnceTheGracePeriodExpired() throws Exception { final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); Thread.sleep(GRACE_PERIOD + 50); assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); @@ -94,14 +104,38 @@ public void gracePeriodOfAShutdownIsNotSpentByAnEarlierMessage() throws Exceptio final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); // an import disables then re-enables the replication service - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); - shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); + final CSN sentByTheImport = newCSN(SERVER_ID, 1); + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheImport); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport); Thread.sleep(GRACE_PERIOD + 50); // the shutdown of the process, much later - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID, 2)); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + /** + * The message of an earlier announcement may still be queued behind a backlog when the + * shutdown announces the replica offline again. Forwarding that older message says nothing + * about the one the shutdown is waiting for, so it must not end the wait. + */ + @Test + public void aStaleForwardDoesNotConsumeTheGracePeriodOfANewerMessage() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN queuedByAnEarlierImport = newCSN(SERVER_ID, 1); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, queuedByAnEarlierImport); + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport); assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); } @Test @@ -109,9 +143,9 @@ public void gracePeriodIsCountedPerDomain() throws Exception { final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); Thread.sleep(GRACE_PERIOD + 50); - shutdownSync.replicaOfflineMsgSent(baseDN2, SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID)); assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); assertThat(shutdownSync.canShutdown(baseDN2)).isFalse(); @@ -124,24 +158,136 @@ public void gracePeriodIsCountedPerDomain() throws Exception @Test public void theForwardOfAnotherReplicasMessageDoesNotEndTheWait() throws Exception { - final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); - shutdownSync.replicaOfflineMsgForwarded(baseDN1, OTHER_SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID)); assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); } + /** The domain waits for the message of every one of its replicas, not for the first of them. */ @Test - public void aReplicaStillWaitingDoesNotHoldBackAnAlreadyForwardedOne() throws Exception + public void aReplicaWhichIsStillWaitingHoldsBackTheShutdownOfItsDomain() throws Exception { - final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofOneReplica = newCSN(SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(OTHER_SERVER_ID)); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } - shutdownSync.replicaOfflineMsgSent(baseDN1, SERVER_ID); - shutdownSync.replicaOfflineMsgSent(baseDN1, OTHER_SERVER_ID); - shutdownSync.replicaOfflineMsgForwarded(baseDN1, SERVER_ID); - shutdownSync.replicaOfflineMsgForwarded(baseDN1, OTHER_SERVER_ID); + @Test + public void canShutdownOnceEveryReplicaOfTheDomainIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofOneReplica = newCSN(SERVER_ID); + final CSN ofTheOtherReplica = newCSN(OTHER_SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheOtherReplica); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica); assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); } + + /** + * The domains of a shutdown wait together: the wait ends when the message of every one of them + * has been forwarded, not when the first one has. Waiting for them one after the other would + * leave the domains which come later without a grace period at all, since the wait of the + * first one spends the deadline they share. + */ + @Test + public void oneWaitCoversEveryDomainOfTheShutdown() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofTheFirstDomain = newCSN(SERVER_ID, 1); + final CSN ofTheSecondDomain = newCSN(SERVER_ID, 2); + shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheFirstDomain); + shutdownSync.replicaOfflineMsgSent(baseDN2, ofTheSecondDomain); + final Thread forwarder = + newForwarderThread(shutdownSync, ofTheFirstDomain, ofTheSecondDomain); + + final long startTime = System.nanoTime(); + forwarder.start(); + shutdownSync.awaitReplicaOfflineMsgsForwarded( + asList(baseDN1, baseDN2), shutdownSync.newShutdownDeadline()); + final long elapsed = millisSince(startTime); + forwarder.join(); + + assertThat(elapsed) + .as("the wait ended on the first domain forwarded, leaving the second one nothing") + .isGreaterThanOrEqualTo(2 * FORWARD_DELAY); + assertThat(elapsed).isLessThan(LONG_GRACE_PERIOD); + } + + /** + * However long the messages of a shutdown may still hold it back, the deadline the shutdown + * was given bounds the wait. + */ + @Test + public void theWaitIsBoundedByTheDeadlineOfTheShutdown() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID)); + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(GRACE_PERIOD); + + final long startTime = System.nanoTime(); + shutdownSync.awaitReplicaOfflineMsgsForwarded(asList(baseDN1, baseDN2), deadline); + final long elapsed = millisSince(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(GRACE_PERIOD - 50); + assertThat(elapsed) + .as("the wait outlived the deadline of the shutdown") + .isLessThan(2 * GRACE_PERIOD); + } + + /** Forwards the message of the first domain, then, as long again later, of the second one. */ + private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, + final CSN ofTheFirstDomain, final CSN ofTheSecondDomain) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + try + { + Thread.sleep(FORWARD_DELAY); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain); + Thread.sleep(FORWARD_DELAY); + shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + }); + } + + private static long millisSince(long startTime) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); + } + + /** + * The CSN of a message a replica announced. They are built by hand rather than with a + * CSNGenerator: this class has no state to share with the server, and a generator would tie + * the test to the time service the server starts. + */ + private static CSN newCSN(int serverId) + { + return newCSN(serverId, 1); + } + + private static CSN newCSN(int serverId, int seqNum) + { + return new CSN(1, seqNum, serverId); + } }