From 2c7e382da347e11820463fcf977d885c2f40d764 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 10 Sep 2026 08:50:34 +0000
Subject: [PATCH] [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg (#947)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java | 9
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java | 19
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java | 660 ++++++++++++++++++++++++++++-
opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java | 224 +++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 205 ++++++--
opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java | 150 ++++++
6 files changed, 1,172 insertions(+), 95 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 ef1ba8b..29a5572 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
@@ -1226,6 +1226,15 @@
* 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.
+ * <p>
+ * The recipients DSRSShutdownSync records when the message is queued are the sharper source of
+ * truth and cover the domains this test lets through - a peer which connected after the
+ * message was queued owes nothing, and is not waited for. What this test still covers is the
+ * announcement which was never queued here at all, and for which no recipient can therefore be
+ * recorded: the message of a replica which picked a remote replication server, or the
+ * announcement of issue #918 recorded after its message was relayed. Those wait out the whole
+ * grace period on the first forward, and on a server with no peer connected nothing would ever
+ * report one.
*/
private void awaitReplicaOfflineMsgsForwarded()
{
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 1674f4c..c402c5b 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
@@ -361,15 +361,23 @@
// Push the message to the replication servers
if (sourceHandler.isDataServer())
{
- for (ReplicationServerHandler rsHandler : connectedRSs.values())
+ if (updateMsg instanceof ReplicaOfflineMsg)
{
- /**
- * Ignore updates to RS with bad gen id
- * (no system managed status for a RS)
- */
- if (!isDifferentGenerationId(rsHandler, updateMsg))
+ pushReplicaOfflineMsgToReplicationServers(
+ updateMsg, notAssuredUpdateMsg, assuredServers);
+ }
+ else
+ {
+ for (ReplicationServerHandler rsHandler : connectedRSs.values())
{
- addUpdate(rsHandler, updateMsg, notAssuredUpdateMsg, assuredServers);
+ /**
+ * Ignore updates to RS with bad gen id
+ * (no system managed status for a RS)
+ */
+ if (!isDifferentGenerationId(rsHandler, updateMsg))
+ {
+ addUpdate(rsHandler, updateMsg, notAssuredUpdateMsg, assuredServers);
+ }
}
}
}
@@ -386,6 +394,54 @@
}
}
+ /**
+ * Pushes a ReplicaOfflineMsg to the replication servers which have to relay it, recording
+ * which of them it goes to before any of them can forward it.
+ * <p>
+ * Each of them is served by its own writer, and the shutdown of a collocated directory server
+ * waits for every one of them to have forwarded the message before it stops the handlers - see
+ * DSRSShutdownSync. The recipients are recorded first because a forward reported before they
+ * are known ends that wait at once.
+ */
+ private void pushReplicaOfflineMsgToReplicationServers(UpdateMsg offlineMsg,
+ NotAssuredUpdateMsg notAssuredUpdateMsg, List<Integer> assuredServers)
+ {
+ final List<ReplicationServerHandler> recipients = new ArrayList<>(connectedRSs.size());
+ final List<Integer> recipientIds = new ArrayList<>(connectedRSs.size());
+ for (ReplicationServerHandler rsHandler : connectedRSs.values())
+ {
+ // Ignore updates to RS with bad gen id (no system managed status for a RS)
+ if (!isDifferentGenerationId(rsHandler, offlineMsg))
+ {
+ recipients.add(rsHandler);
+ recipientIds.add(rsHandler.getServerId());
+ }
+ }
+ localReplicationServer.getDSRSShutdownSync()
+ .replicaOfflineMsgDispatched(baseDN, offlineMsg.getCSN(), recipientIds);
+ for (ReplicationServerHandler rsHandler : recipients)
+ {
+ /*
+ * A recipient whose teardown started while the list was being built has had its message
+ * queue cleared already, and its own give-up ran before it was recorded here: queueing for
+ * it would only leave the shutdown waiting for a forward which can no longer happen.
+ * <p>
+ * This is read after the recipients were recorded, and every teardown - stopServer() and
+ * unregisterFailedHandshake() - raises the flag before it gives up. So of the two, at
+ * least one always sees the other: a teardown whose give-up came too early to find this
+ * recipient has necessarily raised the flag this reads.
+ */
+ if (rsHandler.shuttingDown())
+ {
+ noLongerAwaitTheForwardOf(rsHandler);
+ }
+ else
+ {
+ addUpdate(rsHandler, offlineMsg, notAssuredUpdateMsg, assuredServers);
+ }
+ }
+ }
+
private boolean isDifferentGenerationId(ReplicationServerHandler rsHandler,
UpdateMsg updateMsg)
{
@@ -1068,60 +1124,89 @@
if (!sHandler.engageShutdown())
// Only do this once (prevent other thread to enter here again)
{
- if (!shutdown)
- {
- try
- {
- // Acquire lock on domain (see more details in comment of start()
- // method of ServerHandler)
- lock();
- }
- catch (InterruptedException ex)
- {
- // We can't deal with this here, so re-interrupt thread so that it is
- // caught during subsequent IO.
- Thread.currentThread().interrupt();
- return;
- }
- }
-
try
{
- // Stop useless monitoring publisher if no more RS or DS in domain
- if ( (connectedDSs.size() + connectedRSs.size() )== 1)
- {
- if (logger.isTraceEnabled())
- {
- debug("remote server " + sHandler
- + " is the last RS/DS to be stopped:"
- + " stopping monitoring publisher");
- }
- stopMonitoringPublisher();
- }
-
- if (connectedRSs.containsKey(sHandler.getServerId()))
- {
- unregisterServerHandler(sHandler, shutdown, false);
- }
- else if (connectedDSs.containsKey(sHandler.getServerId()))
- {
- unregisterServerHandler(sHandler, shutdown, true);
- }
- }
- catch(Exception e)
- {
- logger.error(LocalizableMessage.raw(stackTraceToSingleLineString(e)));
- }
- finally
- {
if (!shutdown)
{
- release();
+ try
+ {
+ // Acquire lock on domain (see more details in comment of start()
+ // method of ServerHandler)
+ lock();
+ }
+ catch (InterruptedException ex)
+ {
+ // We can't deal with this here, so re-interrupt thread so that it is
+ // caught during subsequent IO.
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ try
+ {
+ // Stop useless monitoring publisher if no more RS or DS in domain
+ if ( (connectedDSs.size() + connectedRSs.size() )== 1)
+ {
+ if (logger.isTraceEnabled())
+ {
+ debug("remote server " + sHandler
+ + " is the last RS/DS to be stopped:"
+ + " stopping monitoring publisher");
+ }
+ stopMonitoringPublisher();
+ }
+
+ if (connectedRSs.containsKey(sHandler.getServerId()))
+ {
+ unregisterServerHandler(sHandler, shutdown, false);
+ }
+ else if (connectedDSs.containsKey(sHandler.getServerId()))
+ {
+ unregisterServerHandler(sHandler, shutdown, true);
+ }
+ }
+ catch(Exception e)
+ {
+ logger.error(LocalizableMessage.raw(stackTraceToSingleLineString(e)));
+ }
+ finally
+ {
+ if (!shutdown)
+ {
+ release();
+ }
+ }
+ }
+ finally
+ {
+ /*
+ * On every exit of this block, including the interrupted lock acquisition above and an
+ * unregistration which threw: the flag engageShutdown() has just set is one-shot, so no
+ * later stopServer() will run for this handler and nothing else would strike it off.
+ * Every caller closed the session before getting here or is stopping the handler on
+ * purpose, so this peer can no longer be told anything whatever this method managed to
+ * do.
+ */
+ if (sHandler.isReplicationServer())
+ {
+ noLongerAwaitTheForwardOf(sHandler);
}
}
}
}
+ /**
+ * A peer replication server which is gone can no longer forward the ReplicaOfflineMsg it was
+ * given, so the shutdown of a collocated directory server must stop waiting for it - see
+ * DSRSShutdownSync.
+ */
+ void noLongerAwaitTheForwardOf(ServerHandler rsHandler)
+ {
+ localReplicationServer.getDSRSShutdownSync()
+ .replicaOfflineMsgNotForwarded(baseDN, rsHandler.getServerId());
+ }
+
private void unregisterServerHandler(ServerHandler sHandler, boolean shutdown,
boolean isDirectoryServer)
{
@@ -1194,6 +1279,22 @@
{
return;
}
+ /*
+ * This handler is stopping, and nothing else will ever say so for it: the reader and writer
+ * threads whose death normally reaches stopServer() were never started, so no teardown of
+ * its own will run. The flag has to be set before the give-up below, because that is the
+ * order put() reads the two in - it records the recipients of a ReplicaOfflineMsg and only
+ * then asks each of them whether it is shutting down. Skip it and a message being pushed
+ * concurrently slips between the two: the give-up runs while nothing is recorded yet and
+ * does nothing, the recipients are then recorded with this handler among them, and the loop
+ * queues the message for a handler whose queue is about to be cleared - leaving the shutdown
+ * to wait out its whole grace period for a forward nobody can report.
+ */
+ sHandler.engageShutdown();
+ if (!isDataServer)
+ {
+ noLongerAwaitTheForwardOf(sHandler);
+ }
if (connectedDSs.isEmpty() && connectedRSs.isEmpty())
{
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 aa9dc21..9c91cec 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
@@ -105,7 +105,22 @@
"Connection closure: null update returned by domain.");
break;
}
- if (!isUpdateMsgFiltered(updateMsg))
+ if (isUpdateMsgFiltered(updateMsg))
+ {
+ /*
+ * The message is dropped here and will not be published to this server. When it is the
+ * ReplicaOfflineMsg a shutdown is waiting for - its filter is wider than the one
+ * ReplicationServerDomain.put() applied when it queued the message, so a peer RS can
+ * be given a message this drops - the shutdown must stop waiting for a forward which
+ * will never be reported.
+ */
+ if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
+ {
+ dsrsShutdownSync.replicaOfflineMsgNotForwarded(
+ replicationServerDomain.getBaseDN(), handler.getServerId());
+ }
+ }
+ else
{
// Publish the update to the remote server using a protocol version it supports
session.publish(updateMsg);
@@ -121,7 +136,7 @@
if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
{
dsrsShutdownSync.replicaOfflineMsgForwarded(
- replicationServerDomain.getBaseDN(), updateMsg.getCSN());
+ replicationServerDomain.getBaseDN(), updateMsg.getCSN(), handler.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 3d962b0..ba432c4 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
@@ -20,6 +20,8 @@
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import java.util.Collection;
+import java.util.Map.Entry;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -51,9 +53,13 @@
private final long gracePeriod;
/**
- * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per
+ * The ReplicaOfflineMsg which is still owed a forward, per domain and per
* replica of that domain.
* <p>
+ * An entry lives until every replication server the message was queued for
+ * has forwarded it, so it legitimately holds a message some of them have
+ * already sent: what is pending is the forward, not the message.
+ * <p>
* 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
@@ -63,6 +69,10 @@
* 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.
+ * <p>
+ * Each entry knows the replication servers its message was queued for, because each of them is
+ * served by its own writer: the forward of one of them says nothing about the others, whose
+ * queue the shutdown is about to clear.
*/
private final ConcurrentMap<DN, ConcurrentMap<Integer, PendingOfflineMsg>> replicaOfflineMsgs =
new ConcurrentHashMap<>();
@@ -103,14 +113,54 @@
}
/**
- * Message has been forwarded.
+ * Message has been queued for the replication servers which must forward it.
+ * <p>
+ * This must be called before the message is queued for any of them: a replication server can
+ * forward it as soon as it is in its queue, and a forward which finds no recipient recorded
+ * ends the wait at once.
+ *
+ * @param baseDN
+ * the domain for which the message has been sent
+ * @param offlineCSN
+ * the CSN of the message which is being queued
+ * @param replicationServerIds
+ * the server ids of the replication servers the message is being queued for
+ */
+ public void replicaOfflineMsgDispatched(
+ DN baseDN, CSN offlineCSN, Collection<Integer> replicationServerIds)
+ {
+ final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+ if (msgs == null)
+ {
+ return;
+ }
+ final int serverId = offlineCSN.getServerId();
+ final PendingOfflineMsg pending = msgs.get(serverId);
+ /*
+ * The message being queued may be an older announcement of the same replica - one which was
+ * queued behind a backlog since an earlier import. The replication servers it goes to say
+ * nothing about the announcement the shutdown is waiting for.
+ */
+ if (pending != null && pending.csn.equals(offlineCSN)
+ && pending.awaitForwardsFrom(replicationServerIds))
+ {
+ // queued for nobody: there is nothing to wait for
+ msgs.remove(serverId, pending);
+ notifyForwarded();
+ }
+ }
+
+ /**
+ * Message has been forwarded to one of the replication servers it was queued for.
*
* @param baseDN
* the domain for which the message has been sent
* @param forwardedCSN
* the CSN of the forwarded message
+ * @param replicationServerId
+ * the server id of the replication server the message has been forwarded to
*/
- public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN)
+ public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN, int replicationServerId)
{
final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
if (msgs != null)
@@ -124,11 +174,47 @@
* 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))
+ if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN)
+ && pending.forwardedBy(replicationServerId))
{
msgs.remove(serverId, pending);
}
}
+ notifyForwarded();
+ }
+
+ /**
+ * A replication server the message may have been queued for will not forward it: it is gone,
+ * or the message was dropped on its way out.
+ * <p>
+ * Whatever it was given can no longer reach it, so the shutdown must not spend the rest of its
+ * grace period waiting for it.
+ *
+ * @param baseDN
+ * the domain the replication server is connected to
+ * @param replicationServerId
+ * the server id of the replication server which will not forward the message
+ */
+ public void replicaOfflineMsgNotForwarded(DN baseDN, int replicationServerId)
+ {
+ final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+ if (msgs != null)
+ {
+ for (Entry<Integer, PendingOfflineMsg> entry : msgs.entrySet())
+ {
+ final PendingOfflineMsg pending = entry.getValue();
+ if (pending.giveUpOn(replicationServerId))
+ {
+ msgs.remove(entry.getKey(), pending);
+ }
+ }
+ }
+ notifyForwarded();
+ }
+
+ /** Wakes up the shutdown, which re-reads what is left to wait for. */
+ private void notifyForwarded()
+ {
synchronized (forwardedMonitor)
{
forwardedMonitor.notifyAll();
@@ -137,7 +223,8 @@
/**
* Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg
- * has been forwarded or its grace period has expired.
+ * has been forwarded by every replication server it was queued for, or its
+ * grace period has expired.
* <p>
* The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)}
* rather than polling this; it is the same state, observable without waiting for it.
@@ -269,6 +356,11 @@
private final CSN csn;
/** When the message was announced, on the {@link System#nanoTime()} clock. */
private final long sentTime;
+ /**
+ * The replication servers the message was queued for and which have not forwarded it yet,
+ * null as long as it has not been queued for anybody.
+ */
+ private volatile Set<Integer> awaitedForwarders;
private PendingOfflineMsg(CSN csn, long sentTime)
{
@@ -276,10 +368,56 @@
this.sentTime = sentTime;
}
+ /**
+ * Records the replication servers the message is being queued for, and returns whether there
+ * is none of them, i.e. nothing left to wait for.
+ */
+ private boolean awaitForwardsFrom(Collection<Integer> replicationServerIds)
+ {
+ final Set<Integer> awaited = ConcurrentHashMap.newKeySet();
+ awaited.addAll(replicationServerIds);
+ awaitedForwarders = awaited;
+ return awaited.isEmpty();
+ }
+
+ /**
+ * Records the forward of one replication server, and returns whether nothing is left to wait
+ * for.
+ */
+ private boolean forwardedBy(int replicationServerId)
+ {
+ final Set<Integer> awaited = awaitedForwarders;
+ if (awaited == null)
+ {
+ /*
+ * The message never went through the collocated RS - a replica which picked a remote one
+ * announcing itself offline, or an announcement recorded after the message it belongs to
+ * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait
+ * had before the recipients were tracked: the first forward ends it.
+ */
+ return true;
+ }
+ awaited.remove(replicationServerId);
+ return awaited.isEmpty();
+ }
+
+ /**
+ * Gives up on the forward of one replication server, and returns whether nothing is left to
+ * wait for. Unlike a forward, this releases nothing while no recipient is known: a peer going
+ * away says nothing about a message it was never given.
+ */
+ private boolean giveUpOn(int replicationServerId)
+ {
+ final Set<Integer> awaited = awaitedForwarders;
+ return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
+ }
+
@Override
public String toString()
{
- return "PendingOfflineMsg(" + csn + ")";
+ final Set<Integer> awaited = awaitedForwarders;
+ return "PendingOfflineMsg(" + csn
+ + (awaited != null ? ", awaiting the forward of " + awaited : "") + ")";
}
}
}
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 18c7002..e30d13c 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
@@ -19,30 +19,39 @@
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
+import java.util.Collection;
+import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
+import org.opends.server.core.DirectoryServer;
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.DeleteMsg;
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.protocol.WindowMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.StaticUtils;
@@ -57,8 +66,9 @@
* <p>
* 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. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} pins the outcome
- * those waits exist for, on a peer connected through the real handshake.
+ * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} and
+ * {@link #theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline()} pin the outcome those
+ * waits exist for, on peers connected through the real handshake.
*/
@SuppressWarnings("javadoc")
public class ReplicationServerShutdownSyncTest extends ReplicationTestCase
@@ -70,8 +80,25 @@
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;
+ /** The peer replication server whose writer is held back by a full send window. */
+ private static final int HELD_BACK_RS_ID = 95;
+ /** The peer replication server whose handshake is aborted while the message is pushed. */
+ private static final int ABORTED_RS_ID = 96;
+ /** Send window a peer advertises when nothing has to hold its writer back. */
+ private static final int PEER_WINDOW = 100;
+ /**
+ * Send window of the peer which is held back: one change fills it, and the message which
+ * follows stays with its writer until the peer gives it credit again.
+ */
+ private static final int HELD_BACK_PEER_WINDOW = 1;
/** Time given to the forwarding thread before it releases the shutdown. */
private static final long FORWARD_DELAY = 500;
+ /**
+ * Time given to a writer to reach the message it was handed once its send window is opened,
+ * well short of the grace period so that a writer which never gets there is reported as such
+ * rather than as a shutdown which waited.
+ */
+ private static final long WRITER_REACTION_TIMEOUT_MS = 10000;
/** 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;
@@ -170,8 +197,8 @@
final ReplicationServerDomain domain =
replicationServer.getReplicationServerDomain(baseDN, true);
- waitForConnectedReplicationServer(domain);
- final Future<ReplicaOfflineMsg> received = peer.receiveReplicaOfflineMsg();
+ waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
+ final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
/*
* The replica announces itself offline once the shutdown of the replication server is
@@ -189,7 +216,7 @@
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())
+ + "read ended with: %s", peer.failure())
.isNotNull();
assertThat(forwarded.getCSN().getServerId()).isEqualTo(LOCAL_DS_ID);
assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
@@ -266,6 +293,389 @@
}
/**
+ * The grace period covers every peer replication server, not only the fastest of them. The
+ * message is queued for each of them and published by its own writer, and the shutdown clears
+ * the queue and closes the session of whoever has not published it yet: a peer whose writer is
+ * held back would be left unaware that the replica went offline, and its change number indexer
+ * would keep the medium consistency point pinned to the last change of that replica.
+ */
+ @Test
+ public void theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ FakePeerReplicationServer heldBackPeer = null;
+ Thread windowOpener = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer =
+ newReplicationServer(shutdownSync, "shutdownSyncEveryPeerDb", 8229, replicationPort);
+ broker =
+ openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
+ peer = new FakePeerReplicationServer(
+ replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW);
+ heldBackPeer = new FakePeerReplicationServer(
+ replicationPort, HELD_BACK_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);
+
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
+ waitForConnectedReplicationServer(domain, HELD_BACK_RS_ID);
+
+ /*
+ * One change fills the send window of the held back peer: its writer publishes that one and
+ * then blocks on the permit of the next message, so the ReplicaOfflineMsg stays with it
+ * while its neighbour forwards the same message right away.
+ */
+ final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
+ final Future<DeleteMsg> windowFiller = heldBackPeer.receive(DeleteMsg.class);
+ broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "offline-entry-uuid"));
+ assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the send window of the held back peer was never filled, its exchange ended "
+ + "with: %s", heldBackPeer.failure())
+ .isNotNull();
+
+ final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
+ final Future<ReplicaOfflineMsg> receivedWhenHeldBack =
+ heldBackPeer.receive(ReplicaOfflineMsg.class);
+ final CSN offlineCSN = csns.newCSN();
+ shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+ broker.publish(new ReplicaOfflineMsg(offlineCSN));
+ windowOpener = newWindowOpenerThread(heldBackPeer);
+
+ final long startTime = System.nanoTime();
+ windowOpener.start();
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ /*
+ * The barrier first, because it is the one the shutdown is made of and it cannot race:
+ * the wait ends when both writers have reported, or when the grace period runs out, and
+ * the duration below tells the two apart.
+ */
+ assertThat(shutdownSync.forwardedBy())
+ .as("the wait ended on the first peer forwarding the message, without the writer of "
+ + "the peer which was held back ever reporting one")
+ .contains(REMOTE_RS_ID, HELD_BACK_RS_ID);
+ assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
+ .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the peer which was not held back never received the message, its read ended "
+ + "with: %s", peer.failure())
+ .isNotNull();
+ /*
+ * The forward asserted above proves the message reached the Session, not the wire: close()
+ * discards whatever is still in its send queue without draining it, which is the
+ * limitation issue #919 recorded. If this is the only assertion which fails, that window
+ * is the explanation rather than the granularity of the barrier.
+ */
+ assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the peer which was held back never learned that the replica went offline, "
+ + "although its writer reported the forward, its exchange ended with: %s",
+ heldBackPeer.failure())
+ .isNotNull();
+ }
+ finally
+ {
+ joinQuietly(windowOpener);
+ closeQuietly(heldBackPeer);
+ closeQuietly(peer);
+ stop(broker);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A message its writer drops on the way out will never be forwarded, so the shutdown must stop
+ * waiting for the peer it was queued for. The filter of the writer is wider than the one
+ * ReplicationServerDomain.put() applies when it queues the message - it drops anything for a
+ * peer whose generation id is unknown as well - so a peer can be given a message which is then
+ * dropped, and nothing would ever report a forward for it.
+ */
+ @Test
+ public void theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer =
+ newReplicationServer(shutdownSync, "shutdownSyncDroppedMsgDb", 8230, replicationPort);
+ broker =
+ openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
+ peer = new FakePeerReplicationServer(
+ replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);
+
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
+ final ReplicationServerHandler rsHandler = domain.getConnectedRSs().get(REMOTE_RS_ID);
+
+ /*
+ * One change fills the send window of the peer, so the message which follows stays with
+ * its writer until the window is opened again. The writer takes the message off the queue
+ * before it blocks on the permit and evaluates its filter only once it has it, which is
+ * what makes the generation id below take effect on a message already queued.
+ */
+ final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
+ final Future<DeleteMsg> windowFiller = peer.receive(DeleteMsg.class);
+ broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "offline-entry-uuid"));
+ assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the send window of the peer was never filled, its exchange ended with: %s",
+ peer.failure())
+ .isNotNull();
+
+ final CSN offlineCSN = csns.newCSN();
+ shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+ broker.publish(new ReplicaOfflineMsg(offlineCSN));
+ shutdownSync.awaitDispatch();
+ assertThat(shutdownSync.dispatchedTo())
+ .as("the message was never queued for the peer, so its writer has nothing to drop")
+ .contains(REMOTE_RS_ID);
+
+ // the peer no longer shares the generation id of the domain, so its writer drops what was
+ // queued for it before that - a filter wider than the one put() applied
+ rsHandler.setGenerationId(domain.getGenerationId() + 1);
+ peer.openSendWindow(PEER_WINDOW);
+ awaitGiveUpOn(shutdownSync, REMOTE_RS_ID,
+ "the writer dropped the message without telling the shutdown to stop waiting for the "
+ + "peer it was queued for");
+
+ final long startTime = System.nanoTime();
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ assertThat(elapsed)
+ .as("the shutdown kept waiting for a forward its own writer had already dropped")
+ .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ }
+ finally
+ {
+ closeQuietly(peer);
+ stop(broker);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A peer whose handshake is aborted while the message is being pushed must not be waited for.
+ * <p>
+ * put() reads the peers of the domain, records them as the recipients of the message and only
+ * then queues it for each of them. unregisterFailedHandshake() runs on the handshake thread
+ * and takes no domain lock, so it can land in between: its own give-up then finds nothing
+ * recorded yet and does nothing. Nothing else would ever strike that peer off - the reader and
+ * writer threads whose death reaches stopServer() were never started for a handshake which was
+ * aborted - so the shutdown waits out its whole grace period for a forward which cannot come.
+ */
+ @Test
+ public void theShutdownStopsWaitingForAPeerWhoseHandshakeWasAborted() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ try (ServerSocket listen = TestCaseUtils.bindFreePort())
+ {
+ listen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer = newReplicationServer(
+ shutdownSync, "shutdownSyncAbortedHandshakeDb", 8231, 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, REMOTE_RS_ID);
+
+ final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
+ try (Session remoteEnd = sessionPair[0];
+ Session session = sessionPair[1])
+ {
+ // a second peer, registered as a handshake does just before it fails
+ final ReplicationServerHandler aborting = registerConnectedReplicationServer(
+ replicationServer, baseDN, session, ABORTED_RS_ID);
+ aborting.setGenerationId(domain.getGenerationId());
+ shutdownSync.runWhileDispatching(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ domain.unregisterFailedHandshake(aborting);
+ }
+ });
+
+ final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
+ final CSN offlineCSN = newOfflineCSN();
+ shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+ broker.publish(new ReplicaOfflineMsg(offlineCSN));
+ shutdownSync.awaitDispatch();
+
+ final long startTime = System.nanoTime();
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ assertThat(shutdownSync.dispatchedTo())
+ .as("the peer whose handshake was aborted was not recorded among the recipients, so "
+ + "this test never reproduced the window it is about")
+ .contains(REMOTE_RS_ID, ABORTED_RS_ID);
+ assertThat(elapsed)
+ .as("the shutdown waited for a peer whose handshake had been aborted, and which "
+ + "nothing else will ever strike off")
+ .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the peer which was still connected never learned that the replica went "
+ + "offline, its read ended with: %s", peer.failure())
+ .isNotNull();
+ }
+ }
+ finally
+ {
+ closeQuietly(peer);
+ stop(broker);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A peer which disconnects during the grace period can no longer forward what it was given -
+ * its session is closed under its writer - so stopServer() must strike it off rather than let
+ * the shutdown wait out the rest of its window for a peer which is already gone.
+ */
+ @Test
+ public void theShutdownStopsWaitingForAPeerWhichDisconnected() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ FakePeerReplicationServer heldBackPeer = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer = newReplicationServer(
+ shutdownSync, "shutdownSyncDisconnectedPeerDb", 8232, replicationPort);
+ broker =
+ openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
+ peer = new FakePeerReplicationServer(
+ replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW);
+ heldBackPeer = new FakePeerReplicationServer(
+ replicationPort, HELD_BACK_RS_ID, baseDN, EMPTY_DN_GENID, HELD_BACK_PEER_WINDOW);
+
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
+ waitForConnectedReplicationServer(domain, HELD_BACK_RS_ID);
+
+ // the peer which disconnects is held back by a full send window, so that it cannot have
+ // forwarded the message before it goes: what ends the wait must be the give-up
+ final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
+ final Future<DeleteMsg> windowFiller = heldBackPeer.receive(DeleteMsg.class);
+ broker.publish(new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "offline-entry-uuid"));
+ assertThat(windowFiller.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the send window of the held back peer was never filled, its exchange ended "
+ + "with: %s", heldBackPeer.failure())
+ .isNotNull();
+
+ final Future<ReplicaOfflineMsg> received = peer.receive(ReplicaOfflineMsg.class);
+ final CSN offlineCSN = csns.newCSN();
+ shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+ broker.publish(new ReplicaOfflineMsg(offlineCSN));
+ shutdownSync.awaitDispatch();
+ heldBackPeer.close();
+
+ final long startTime = System.nanoTime();
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ assertThat(shutdownSync.dispatchedTo())
+ .as("the message was not queued for the peer which then disconnected, so this test "
+ + "never reproduced what it is about")
+ .contains(REMOTE_RS_ID, HELD_BACK_RS_ID);
+ assertThat(elapsed)
+ .as("the shutdown kept waiting for a forward from a peer which had disconnected")
+ .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ assertThat(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("the peer which stayed connected never received the message, its read ended "
+ + "with: %s", peer.failure())
+ .isNotNull();
+ }
+ finally
+ {
+ closeQuietly(heldBackPeer);
+ closeQuietly(peer);
+ stop(broker);
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * A peer which does not share the generation id of the domain is not given the message, so it
+ * must not be recorded among the peers the shutdown waits for. The caller side check on
+ * getConnectedRSs() does not cover this: such a peer is connected, and would make the domain
+ * spend its grace period on a message it was never queued.
+ */
+ @Test
+ public void theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer = newReplicationServer(
+ shutdownSync, "shutdownSyncOtherGenerationIdDb", 8233, 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, REMOTE_RS_ID);
+ domain.getConnectedRSs().get(REMOTE_RS_ID).setGenerationId(domain.getGenerationId() + 1);
+
+ final CSN offlineCSN = newOfflineCSN();
+ shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+ broker.publish(new ReplicaOfflineMsg(offlineCSN));
+ shutdownSync.awaitDispatch();
+
+ final long startTime = System.nanoTime();
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ assertThat(shutdownSync.dispatchedTo())
+ .as("the message was recorded as queued for a peer which does not share the "
+ + "generation id of the domain, and which it was never queued for")
+ .isEmpty();
+ assertThat(elapsed)
+ .as("the shutdown waited for a peer the message was not queued for")
+ .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ }
+ finally
+ {
+ closeQuietly(peer);
+ stop(broker);
+ 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.
*/
@@ -311,6 +721,9 @@
replicationServer =
newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort);
broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
+ // the writer this test is about exists only once the handshake of the directory server is
+ // over, and the shutdown below would otherwise be free to abort that handshake instead
+ waitForConnectedDirectoryServer(replicationServer.getReplicationServerDomain(baseDN, true));
final long startTime = System.nanoTime();
shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
@@ -434,13 +847,20 @@
* 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(
+ private ReplicationServerHandler registerConnectedReplicationServer(
ReplicationServer replicationServer, DN baseDN, Session session) throws Exception
{
+ return registerConnectedReplicationServer(replicationServer, baseDN, session, REMOTE_RS_ID);
+ }
+
+ private ReplicationServerHandler registerConnectedReplicationServer(
+ ReplicationServer replicationServer, DN baseDN, Session session, int serverId)
+ throws Exception
+ {
final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true);
final ReplicationServerHandler rsHandler =
new ReplicationServerHandler(session, 100, replicationServer, 100);
- rsHandler.serverId = REMOTE_RS_ID;
+ rsHandler.serverId = serverId;
rsHandler.serverURL = "127.0.0.1:1636";
rsHandler.setBaseDNAndDomain(baseDN, false);
domain.lock();
@@ -452,22 +872,68 @@
{
domain.release();
}
+ return rsHandler;
}
- private void waitForConnectedReplicationServer(final ReplicationServerDomain domain)
- throws Exception
+ /** Waits for the barrier to have been told that this peer will not forward the message. */
+ private void awaitGiveUpOn(final RecordingShutdownSync shutdownSync, final int serverId,
+ final String reason) throws Exception
+ {
+ new TestTimer.Builder()
+ .maxSleep(WRITER_REACTION_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(new TestTimer.CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertThat(shutdownSync.gaveUpOn()).as(reason).contains(serverId);
+ }
+ });
+ }
+
+ /**
+ * Waits for the peer replication server to be connected <em>and</em> for its handshake to be
+ * over.
+ * <p>
+ * The registration is not the end of the handshake: {@code startFromRemoteRS()} puts the
+ * handler in {@code connectedRSs} before it calls {@code finalizeStart()}, which is what
+ * starts the reader and the writer, and the whole handshake runs in the listen thread of the
+ * replication server. {@link ReplicationServer#shutdown()} interrupts that thread before it
+ * waits for the ReplicaOfflineMsgs, so a shutdown triggered while the handshake is still in
+ * {@code Session.waitForStartup()} aborts it: the session is closed and the handler
+ * unregistered, and the peer is gone before the message could be queued for it, let alone
+ * forwarded. Issue #821 recorded that same window, from the dead handler it used to leave
+ * behind.
+ * <p>
+ * The listen thread serves one handshake at a time, so a peer which is past that window also
+ * puts every connection accepted before it - the collocated directory server of these tests
+ * among them - past it.
+ */
+ private void waitForConnectedReplicationServer(
+ final ReplicationServerDomain domain, final int serverId) 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);
+ final ReplicationServerHandler rsHandler = domain.getConnectedRSs().get(serverId);
+ assertThat(rsHandler)
+ .as("the peer replication server %s never connected", serverId).isNotNull();
+ assertThat(handshakeIsOver(rsHandler))
+ .as("the handshake of the peer replication server %s never finished", serverId)
+ .isTrue();
}
});
}
+ /**
+ * Waits for the collocated directory server to be connected and for its handshake to be over -
+ * {@link #waitForConnectedReplicationServer(ReplicationServerDomain, int)} says what the
+ * registration alone leaves open.
+ */
private DataServerHandler waitForConnectedDirectoryServer(final ReplicationServerDomain domain)
throws Exception
{
@@ -478,11 +944,24 @@
{
final DataServerHandler dsHandler = domain.getConnectedDSs().get(REMOTE_DS_ID);
assertThat(dsHandler).as("the directory server never connected").isNotNull();
+ assertThat(handshakeIsOver(dsHandler))
+ .as("the handshake of the directory server never finished").isTrue();
return dsHandler;
}
});
}
+ /**
+ * Whether the handshake of the handler is over, so that the interrupt
+ * {@link ReplicationServer#shutdown()} sends to its listen thread can no longer abort it:
+ * {@code ServerHandler.finalizeStart()} registers the handler as a monitor provider by its
+ * last statement, after the reader and the writer have been started.
+ */
+ private static boolean handshakeIsOver(ServerHandler handler)
+ {
+ return DirectoryServer.getMonitorProviders().containsValue(handler);
+ }
+
private static TestTimer newConnectionTimer()
{
return new TestTimer.Builder()
@@ -503,7 +982,7 @@
{
return;
}
- shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN, REMOTE_RS_ID);
}
});
}
@@ -524,6 +1003,26 @@
});
}
+ /**
+ * Gives the held back peer credit to receive again, once the shutdown has had time to end on
+ * the forward of the peer which was not held back.
+ */
+ private Thread newWindowOpenerThread(final FakePeerReplicationServer heldBackPeer)
+ {
+ return new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ if (!sleepQuietly(FORWARD_DELAY))
+ {
+ return;
+ }
+ heldBackPeer.openSendWindow(PEER_WINDOW);
+ }
+ });
+ }
+
private Thread newReAnnouncerThread(final DSRSShutdownSync shutdownSync, final DN baseDN1,
final DN baseDN2, final AtomicBoolean stopped)
{
@@ -691,6 +1190,85 @@
}
/**
+ * A synchronization object which records what the production code reports to it, so that a
+ * test can assert on the barrier the shutdown is made of rather than only on how long it took,
+ * and can interleave a teardown with the dispatch of a message.
+ */
+ private static final class RecordingShutdownSync extends DSRSShutdownSync
+ {
+ private final List<Integer> dispatchedTo = new CopyOnWriteArrayList<>();
+ private final List<Integer> forwardedBy = new CopyOnWriteArrayList<>();
+ private final List<Integer> gaveUpOn = new CopyOnWriteArrayList<>();
+ private final CountDownLatch dispatched = new CountDownLatch(1);
+ /** Runs inside the next dispatch, before the recipients are recorded. */
+ private final AtomicReference<Runnable> whileDispatching = new AtomicReference<>();
+
+ /**
+ * Runs the provided action inside the next dispatch, before the recipients are recorded:
+ * the window in which put() has read the peers of the domain and nothing knows yet which of
+ * them the message is for.
+ */
+ void runWhileDispatching(Runnable action)
+ {
+ whileDispatching.set(action);
+ }
+
+ @Override
+ public void replicaOfflineMsgDispatched(
+ DN baseDN, CSN offlineCSN, Collection<Integer> replicationServerIds)
+ {
+ final Runnable action = whileDispatching.getAndSet(null);
+ if (action != null)
+ {
+ action.run();
+ }
+ dispatchedTo.addAll(replicationServerIds);
+ super.replicaOfflineMsgDispatched(baseDN, offlineCSN, replicationServerIds);
+ dispatched.countDown();
+ }
+
+ @Override
+ public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN, int replicationServerId)
+ {
+ forwardedBy.add(replicationServerId);
+ super.replicaOfflineMsgForwarded(baseDN, forwardedCSN, replicationServerId);
+ }
+
+ @Override
+ public void replicaOfflineMsgNotForwarded(DN baseDN, int replicationServerId)
+ {
+ gaveUpOn.add(replicationServerId);
+ super.replicaOfflineMsgNotForwarded(baseDN, replicationServerId);
+ }
+
+ /** Waits for put() to have pushed a ReplicaOfflineMsg to the peers of its domain. */
+ void awaitDispatch() throws InterruptedException
+ {
+ assertThat(dispatched.await(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
+ .as("no ReplicaOfflineMsg was ever pushed to the peers of the domain")
+ .isTrue();
+ }
+
+ /** The peers the message was recorded as queued for. */
+ List<Integer> dispatchedTo()
+ {
+ return dispatchedTo;
+ }
+
+ /** The peers whose writer reported having forwarded the message. */
+ List<Integer> forwardedBy()
+ {
+ return forwardedBy;
+ }
+
+ /** The peers the shutdown was told to stop waiting for. */
+ List<Integer> gaveUpOn()
+ {
+ return gaveUpOn;
+ }
+ }
+
+ /**
* 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.
@@ -699,12 +1277,23 @@
{
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;
+ /**
+ * What ended the exchange with the replication server, so that a message which never arrived
+ * can be told from an exchange which failed. The reader and the thread which opens the send
+ * window both report here, and the first failure is the one kept: it is the one which
+ * explains the rest.
+ */
+ private final AtomicReference<Exception> failure = new AtomicReference<>();
FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId)
throws Exception
{
+ this(replicationPort, serverId, baseDN, generationId, PEER_WINDOW);
+ }
+
+ FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId,
+ int windowSize) throws Exception
+ {
final Socket socket = new Socket();
Session newSession = null;
boolean handshaken = false;
@@ -716,7 +1305,7 @@
final String serverURL = "127.0.0.1:" + socket.getLocalPort();
final byte groupId = (byte) 1;
- newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, 100,
+ newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, windowSize,
new ServerState(), generationId, false, groupId, 5000));
final ReplServerStartMsg inStartMsg =
waitForSpecificMsg(newSession, ReplServerStartMsg.class);
@@ -748,22 +1337,25 @@
session = newSession;
}
- /** Returns the first ReplicaOfflineMsg this peer receives, or null if its session ends first. */
- Future<ReplicaOfflineMsg> receiveReplicaOfflineMsg()
+ /**
+ * Returns the first message of the given type this peer receives, or null if its session
+ * ends first.
+ */
+ <T extends ReplicationMsg> Future<T> receive(final Class<T> msgClass)
{
- return reader.submit(new Callable<ReplicaOfflineMsg>()
+ return reader.submit(new Callable<T>()
{
@Override
- public ReplicaOfflineMsg call()
+ public T call()
{
try
{
while (true)
{
final ReplicationMsg msg = session.receive();
- if (msg instanceof ReplicaOfflineMsg)
+ if (msgClass.isInstance(msg))
{
- return (ReplicaOfflineMsg) msg;
+ return msgClass.cast(msg);
}
}
}
@@ -771,17 +1363,35 @@
{
// The session is closed when the replication server completes its shutdown: whatever
// has not arrived by then never will.
- readerFailure = e;
+ failed(e);
return null;
}
}
});
}
- /** Returns what ended the read of this peer, null if nothing did. */
- Exception readerFailure()
+ /** Gives the replication server credit to publish again, as a peer which keeps up does. */
+ void openSendWindow(int credits)
{
- return readerFailure;
+ try
+ {
+ session.publish(new WindowMsg(credits));
+ }
+ catch (IOException e)
+ {
+ failed(e);
+ }
+ }
+
+ private void failed(Exception e)
+ {
+ failure.compareAndSet(null, e);
+ }
+
+ /** Returns what ended the exchange with this peer, null if nothing did. */
+ Exception failure()
+ {
+ return failure.get();
}
void 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 81408a0..be482e1 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
@@ -18,6 +18,7 @@
import static java.util.Arrays.asList;
import static org.assertj.core.api.Assertions.assertThat;
+import java.util.Collections;
import java.util.concurrent.TimeUnit;
import org.forgerock.opendj.ldap.DN;
@@ -41,6 +42,9 @@
private static final long FORWARD_DELAY = 200;
private static final int SERVER_ID = 1;
private static final int OTHER_SERVER_ID = 2;
+ /** A peer replication server the collocated one relays the message to. */
+ private static final int RS_ID = 11;
+ private static final int OTHER_RS_ID = 12;
private static DN baseDN1;
private static DN baseDN2;
@@ -77,7 +81,7 @@
final CSN offlineCSN = newCSN(SERVER_ID);
shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
}
@@ -106,7 +110,7 @@
// an import disables then re-enables the replication service
final CSN sentByTheImport = newCSN(SERVER_ID, 1);
shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheImport);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport, RS_ID);
Thread.sleep(GRACE_PERIOD + 50);
// the shutdown of the process, much later
@@ -129,11 +133,11 @@
shutdownSync.replicaOfflineMsgSent(baseDN1, queuedByAnEarlierImport);
shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport, RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
}
@@ -161,7 +165,7 @@
final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID), RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
}
@@ -175,7 +179,7 @@
shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(OTHER_SERVER_ID));
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica, RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
}
@@ -189,8 +193,8 @@
shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheOtherReplica);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica, RS_ID);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica, RS_ID);
assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
}
@@ -247,6 +251,206 @@
.isLessThan(2 * GRACE_PERIOD);
}
+ /**
+ * The collocated replication server queues the message for every peer it relays to, and each
+ * of them is served by its own writer: the forward of one peer says nothing about the others,
+ * whose queue the shutdown is about to clear.
+ */
+ @Test
+ public void theForwardOfOnePeerDoesNotEndTheWaitOfTheOthers() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
+ }
+
+ @Test
+ public void canShutdownOnceEveryPeerTheMessageWasQueuedForForwardedIt() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, OTHER_RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * With no peer to relay the message to - none connected, or none sharing the generation id of
+ * the domain - there is nothing to wait for.
+ */
+ @Test
+ public void canShutdownWhenTheMessageWasQueuedForNoPeer() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(
+ baseDN1, offlineCSN, Collections.<Integer> emptyList());
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * A peer which is no longer connected cannot forward anything, so the shutdown must not spend
+ * the rest of its window waiting for it.
+ */
+ @Test
+ public void aPeerWhichStoppedIsNoLongerWaitedFor() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
+ shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, OTHER_RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /** The peers which are still connected keep their part of the grace period. */
+ @Test
+ public void aPeerWhichStoppedDoesNotEndTheWaitOfTheOthers() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, OTHER_RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
+ }
+
+ /**
+ * A peer which connected after the message was queued was never given it, so what it forwards
+ * is a message of its own catch-up and says nothing about the peers which still owe theirs.
+ */
+ @Test
+ public void theForwardOfAPeerTheMessageWasNotQueuedForDoesNotEndTheWait() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, OTHER_RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
+ }
+
+ /**
+ * The peers are recorded by the collocated replication server when it queues the message for
+ * them, which the message of a replica connected to a remote replication server never reaches.
+ * With no peer recorded the wait keeps the behaviour it had before they were tracked: the
+ * first forward ends it.
+ */
+ @Test
+ public void theFirstForwardEndsTheWaitWhenNoPeerWasRecorded() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN, RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * A peer going away says nothing about a message it was never given, which is the opposite of
+ * what a forward says: with no peer recorded the first forward ends the wait, and a give-up
+ * must leave it running. Otherwise any peer disconnecting would release a message the
+ * collocated replication server has not queued for anybody yet - the very bug the recipients
+ * were introduced to close, in a new shape.
+ */
+ @Test
+ public void aPeerStoppingBeforeTheMessageIsQueuedDoesNotEndTheWait() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+
+ shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
+ shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
+ }
+
+ /**
+ * A replica announces itself offline on every disableService(), so the peers recorded for an
+ * earlier announcement say nothing about the one the shutdown is waiting for.
+ */
+ @Test
+ public void thePeersOfAnEarlierAnnouncementAreNotTakenForThoseOfThisOne() 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, sentByTheShutdown);
+ shutdownSync.replicaOfflineMsgDispatched(
+ baseDN1, queuedByAnEarlierImport, asList(RS_ID, OTHER_RS_ID));
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
+
+ assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
+ }
+
+ /**
+ * The peer which goes away must wake the shutdown up, and not leave it waiting for a forward
+ * nobody can report any more.
+ */
+ @Test
+ public void theWaitEndsWhenTheLastPeerExpectedToForwardStops() throws Exception
+ {
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
+ final CSN offlineCSN = newCSN(SERVER_ID);
+ shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
+ shutdownSync.replicaOfflineMsgDispatched(baseDN1, offlineCSN, asList(RS_ID));
+ final Thread peerStopper = newPeerStopperThread(shutdownSync, RS_ID);
+
+ final long startTime = System.nanoTime();
+ peerStopper.start();
+ shutdownSync.awaitReplicaOfflineMsgsForwarded(
+ asList(baseDN1), shutdownSync.newShutdownDeadline());
+ final long elapsed = millisSince(startTime);
+ peerStopper.join();
+
+ assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY);
+ assertThat(elapsed)
+ .as("the peer going away did not wake the wait up")
+ .isLessThan(LONG_GRACE_PERIOD);
+ }
+
+ /** Stops the peer the message was queued for, as a disconnection during the wait does. */
+ private Thread newPeerStopperThread(final DSRSShutdownSync shutdownSync, final int peerId)
+ {
+ return new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ Thread.sleep(FORWARD_DELAY);
+ shutdownSync.replicaOfflineMsgNotForwarded(baseDN1, peerId);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ }
+
/** 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)
@@ -259,9 +463,9 @@
try
{
Thread.sleep(FORWARD_DELAY);
- shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain, RS_ID);
Thread.sleep(FORWARD_DELAY);
- shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain);
+ shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain, RS_ID);
}
catch (InterruptedException e)
{
--
Gitblit v1.10.0