+ * What this reads, {@link #restartSession(boolean)} claims: a total update the listener + * thread is about to claim is not visible to a read, and the restart must not stop the + * session such a total update reads (issue #1041). This is the early exit of the roads + * which lead to that restart, and the answer for the ones which never restart anything. + *
* The total update owns the session from the moment it is asked for, not from the
* moment its entries stream: the {@code InitializeTargetMsg} which answers the request
* arrives over that session, so a restart made while it is on its way loses it, and the
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 472f68cf8d..80c6bbf5bc 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -56,6 +56,7 @@
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType;
import org.forgerock.opendj.server.config.server.ReplicationDomainCfg;
+import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.api.DirectoryThread;
import org.opends.server.api.MonitorData;
import org.opends.server.backends.task.Task;
@@ -262,6 +263,39 @@ public ECLIncludes addIncludedAttributes(int serverId,
* Null when none is being processed.
*/
private final AtomicReference
+ * A restart of the session reads whether such a total update owns it before it stops
+ * anything, and the listener thread claims the context for the {@code InitializeTargetMsg}
+ * it took off the session - and the two share no lock: {@link #disableService()} waits for
+ * the listener thread under {@link #serviceStateLock}, so the listener can not take that
+ * lock before its claim. A restart which read no owner a few statements before the claim
+ * landed stopped the broker the import was about to read, and the import ended on the
+ * nothing which arrived - recorded as a failed import since issue #1039, over a suffix
+ * which has been replaced by it all the same. The two contend on the one reference
+ * instead: the restart claims it with this context, the claim of the listener fails
+ * against it, and exactly one of them wins - the total update is either the owner the
+ * restart reads, or refused.
+ *
+ * It is neither an import nor an export: {@link #ieRunning()}, {@link #importInProgress()}
+ * and {@link #getImportExportContext()} do not report it.
+ */
+ private static final ImportExportContext SESSION_BEING_STOPPED = new ImportExportContext(false);
+ /**
+ * Run by the listener thread between the {@code InitializeTargetMsg} it took off the
+ * session and its claim of the import context - or, for a total update this replica asked
+ * for, its read of the context the request claimed. Only there for the tests, which hold
+ * the listener thread there: nothing else runs in that gap.
+ */
+ private volatile Runnable importClaimHook;
+ /**
+ * Run by {@link #disableService()} under its locks, before the broker is stopped. Only
+ * there for the tests, which hold a stop of the service there: what the claim of a session
+ * stop is for is the total update which lands between the decision to stop and the stop.
+ */
+ private volatile Runnable serviceStopHook;
/**
* The Thread waiting for incoming update messages for this domain and pushing
@@ -838,7 +872,7 @@ else if (msg instanceof InitializeTargetMsg)
else if (msg instanceof ErrorMsg)
{
ErrorMsg errorMsg = (ErrorMsg)msg;
- ImportExportContext ieCtx = importExportContext.get();
+ ImportExportContext ieCtx = getImportExportContext();
if (ieCtx != null)
{
/*
@@ -900,7 +934,7 @@ else if (msg instanceof UpdateMsg)
}
else if (msg instanceof InitializeRcvAckMsg)
{
- ImportExportContext ieCtx = importExportContext.get();
+ ImportExportContext ieCtx = getImportExportContext();
if (ieCtx != null)
{
InitializeRcvAckMsg ackMsg = (InitializeRcvAckMsg) msg;
@@ -1660,7 +1694,7 @@ context is live here (ErrorMsg carries no correlation id) - and
// Release the context whatever the outcome, otherwise ieRunning() would
// remain true forever (resolves the historical "FIXME should not this
// be in a finally?").
- releaseIEContext();
+ releaseIEContext(ieCtx);
}
}
@@ -2036,16 +2070,27 @@ private ImportExportContext acquireIEContext(boolean importInProgress)
final ImportExportContext ieCtx = new ImportExportContext(importInProgress);
if (!importExportContext.compareAndSet(null, ieCtx))
{
- // Rejects 2 simultaneous exports
- LocalizableMessage message = ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get();
+ // Rejects 2 simultaneous exports, and a total update which is claimed while the
+ // session is being stopped - in either direction: the entries of an export out of this
+ // server are streamed over that session too (see SESSION_BEING_STOPPED)
+ final LocalizableMessage message = importExportContext.get() == SESSION_BEING_STOPPED
+ ? ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId())
+ : ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get();
throw new DirectoryException(ResultCode.OTHER, message);
}
return ieCtx;
}
- private void releaseIEContext()
+ /**
+ * Releases the provided import/export context, and only that one: a road which failed to
+ * acquire a context of its own must not release the one it failed against - the import or
+ * export which owns it, or the claim of a session stop ({@code SESSION_BEING_STOPPED}).
+ *
+ * @param ieCtx the context to release
+ */
+ private void releaseIEContext(ImportExportContext ieCtx)
{
- importExportContext.set(null);
+ importExportContext.compareAndSet(ieCtx, null);
}
/**
@@ -2059,7 +2104,7 @@ private void releaseIEContext()
*/
private void completeInitializeTask(ImportExportContext ieCtx)
{
- releaseIEContext();
+ releaseIEContext(ieCtx);
if (ieCtx.initializeTask instanceof InitializeTask)
{
// Update the task that initiated the import
@@ -2108,7 +2153,7 @@ protected byte[] receiveEntryBytes()
ReplicationMsg msg;
while (true)
{
- ImportExportContext ieCtx = importExportContext.get();
+ ImportExportContext ieCtx = getImportExportContext();
try
{
// In the context of the total update, we don't want any automatic
@@ -2281,7 +2326,7 @@ void exportLDIFEntry(byte[] lDIFEntry, int pos, int length)
}
// build the message
- ImportExportContext ieCtx = importExportContext.get();
+ ImportExportContext ieCtx = getImportExportContext();
EntryMsg entryMessage = new EntryMsg(
getServerId(), ieCtx.getExportTarget(), lDIFEntry, pos, length,
++ieCtx.msgCnt);
@@ -2425,6 +2470,7 @@ running this method directly coming from initialize() method and did
not processed any topology message in between the failure and the
new attempt.
*/
+ ImportExportContext ieCtx = null;
try
{
/*
@@ -2434,7 +2480,7 @@ The context will be used when we (the listener thread) will receive
update the task.
*/
- final ImportExportContext ieCtx = acquireIEContext(true);
+ ieCtx = acquireIEContext(true);
ieCtx.initializeTask = initTask;
ieCtx.attemptCnt = 0;
ieCtx.initReqMsgSent = new InitializeRequestMsg(
@@ -2474,7 +2520,10 @@ The context will be used when we (the listener thread) will receive
{
// No need to call here updateTaskCompletionState - will be done
// by the caller
- releaseIEContext();
+ if (ieCtx != null)
+ {
+ releaseIEContext(ieCtx);
+ }
throw new DirectoryException(ResultCode.OTHER, errMsg);
}
}
@@ -2494,7 +2543,7 @@ The context will be used when we (the listener thread) will receive
*/
public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs)
{
- final ImportExportContext ieCtx = importExportContext.get();
+ final ImportExportContext ieCtx = getImportExportContext();
if (ieCtx == null || !ieCtx.importInProgress() || ieCtx.initReqMsgSent == null
|| !ieCtx.abandonIfStalled(stalledTimeoutMs))
{
@@ -2510,6 +2559,23 @@ public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs)
return true;
}
+ /**
+ * Refuses a total update another server started into this replica: the exporter is told
+ * so that it does not stream to a replica which will discard the entries, and this server
+ * records why the total update it was the target of did not run - the exporter's task
+ * reports the failure, and an administrator reading this server's log has to find it here.
+ *
+ * @param requesterServerId the server which asked for the total update
+ * @param reason why it is refused
+ */
+ private void rejectInitializeTarget(int requesterServerId, LocalizableMessage reason)
+ {
+ logger.error(reason);
+ // Silently not sent over a session which is already stopped: the replication server
+ // then tells the exporter that this replica is not there to stream to.
+ broker.publish(new ErrorMsg(requesterServerId, reason));
+ }
+
/**
* Processes an InitializeTargetMsg received from a remote server
* meaning processes an initialization from the entries expected to be
@@ -2532,9 +2598,14 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester
InitializeTask initFromTask = null;
final int source = initTargetMsgReceived.getSenderID();
final ImportExportContext ieCtx;
+ final Runnable hook = importClaimHook;
+ if (hook != null)
+ {
+ hook.run();
+ }
if (initTargetMsgReceived.getInitiatorID() == getServerId())
{
- ieCtx = importExportContext.get();
+ ieCtx = getImportExportContext();
if (ieCtx == null || !ieCtx.markInitStartReceived())
{
/*
@@ -2551,6 +2622,20 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester
}
return;
}
+ if (broker.shuttingDown())
+ {
+ /*
+ * The same read as for a total update another server started (see below), with the
+ * same window: the context is the one the request claimed, and no restart stops the
+ * session under it - an import owns the session - but the domain going down or being
+ * disabled does. The task which asked for the total update is failed with the reason;
+ * the exporter learns of the stop the way it does of any other stop of this session.
+ */
+ ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
+ ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId())));
+ completeInitializeTask(ieCtx);
+ return;
+ }
}
else
{
@@ -2564,11 +2649,32 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester
}
catch (DirectoryException e)
{
- // A concurrent import/export owns the context: reject this
- // initialization without touching that operation's context, and let
- // the exporter know so that it does not export to a replica that
- // will discard the entries
- broker.publish(new ErrorMsg(requesterServerId, e.getMessageObject()));
+ // A concurrent import/export owns the context, or the session is being stopped:
+ // reject this initialization without touching that operation's context, and let
+ // the exporter know so that it does not export to a replica that will discard the
+ // entries
+ rejectInitializeTarget(requesterServerId, e.getMessageObject());
+ return;
+ }
+ if (broker.shuttingDown())
+ {
+ /*
+ * The claim won against no restart, and the session is being stopped all the same:
+ * the domain is going down or being disabled, or a restart found an export in the
+ * context and stopped the session it streams over. The import would read that broker
+ * as the end of its stream, and what runs before it publishes over the session: it
+ * is refused here, before the backend is taken away.
+ *
+ * Read rather than claimed against: none of these roads claims anything - they stop
+ * the session whatever owns it - so nothing orders this read against the stop. It
+ * narrows the window, it does not close it: a stop which lands after it still has
+ * the import run over a session which is going down, and end as a failed import over
+ * the suffix it has replaced (issue #1039). Every one of these roads had that window
+ * before this claim, and has it still.
+ */
+ releaseIEContext(ieCtx);
+ rejectInitializeTarget(requesterServerId,
+ ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId()));
return;
}
}
@@ -2773,7 +2879,37 @@ private void setNewStatus(StatusMachineEvent event)
*/
public boolean ieRunning()
{
- return importExportContext.get() != null;
+ return getImportExportContext() != null;
+ }
+
+ /**
+ * Sets what the listener thread runs between the {@code InitializeTargetMsg} it took off
+ * the session and its claim of the import context - or, for a total update this replica
+ * asked for, its read of the context the request claimed.
+ *
+ * Only there for the tests which drive something else through that gap: it is a few
+ * statements wide, and nothing else can hold the listener thread there.
+ *
+ * @param hook what to run there, or {@code null} to run nothing
+ */
+ @VisibleForTesting
+ public void setImportClaimHook(Runnable hook)
+ {
+ importClaimHook = hook;
+ }
+
+ /**
+ * Sets what {@link #disableService()} runs, under its locks, before it stops the broker.
+ *
+ * Only there for the tests which drive something else through that gap: a total update
+ * which is claimed after the decision to stop the service and before the stop.
+ *
+ * @param hook what to run there, or {@code null} to run nothing
+ */
+ @VisibleForTesting
+ public void setServiceStopHook(Runnable hook)
+ {
+ serviceStopHook = hook;
}
/**
@@ -2790,7 +2926,7 @@ public boolean ieRunning()
*/
protected boolean importInProgress()
{
- final ImportExportContext ieCtx = importExportContext.get();
+ final ImportExportContext ieCtx = getImportExportContext();
return ieCtx != null && ieCtx.importInProgress();
}
@@ -3343,6 +3479,11 @@ public final void disableService()
{
synchronized (sessionLock)
{
+ final Runnable hook = serviceStopHook;
+ if (hook != null)
+ {
+ hook.run();
+ }
/*
* Stop the broker first in order to prevent the listener from reconnecting - see OPENDJ-457.
*/
@@ -3370,6 +3511,57 @@ public final void disableService()
}
}
+ /**
+ * Stops the Replication Service the way {@link #disableService()} does, unless a total
+ * update into this replica owns the session.
+ *
+ * Whether one does is claimed rather than read (issue #1041): the listener thread claims
+ * the import context for an {@code InitializeTargetMsg} under no lock, so a read of it
+ * under {@link #serviceStateLock} orders nothing. The claim is
+ * {@code SESSION_BEING_STOPPED}, held for the length of the stop and released once the
+ * listener thread is gone - it is the one thread which claims a total update this replica
+ * did not ask for, and {@link #disableService()} waits for it. An export in the context is
+ * not an owner: the session is stopped from under it and the exporter reports the cut, as
+ * it does for every other stop. A total update which lands between the end of that export
+ * and the stop is refused by the listener when it reads the broker as stopping after its
+ * claim; a stop which lands after that read still has the import run over a session which
+ * is going down, and end as a failed import over the suffix it has replaced (issue
+ * #1039): the read narrows that window, it does not close it.
+ *
+ * @return {@code true} when the service was stopped, {@code false} when a total update
+ * into this replica owns the session and it was left alone
+ */
+ protected final boolean disableServiceUnlessImportInProgress()
+ {
+ synchronized (serviceStateLock)
+ {
+ while (!importExportContext.compareAndSet(null, SESSION_BEING_STOPPED))
+ {
+ final ImportExportContext owner = importExportContext.get();
+ if (owner == null)
+ {
+ // Released between the two reads: claim again.
+ continue;
+ }
+ if (owner.importInProgress())
+ {
+ return false;
+ }
+ disableService();
+ return true;
+ }
+ try
+ {
+ disableService();
+ }
+ finally
+ {
+ importExportContext.compareAndSet(SESSION_BEING_STOPPED, null);
+ }
+ return true;
+ }
+ }
+
/**
* Returns {@code true} if the listener thread is shutting down or has
* shutdown.
@@ -3863,7 +4055,8 @@ public void addAdditionalMonitoring(MonitorData monitorData)
*/
protected ImportExportContext getImportExportContext()
{
- return importExportContext.get();
+ final ImportExportContext ieCtx = importExportContext.get();
+ return ieCtx != SESSION_BEING_STOPPED ? ieCtx : null;
}
/**
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index a4eb51f839..0276c4e193 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -712,3 +712,6 @@ ERR_INIT_SESSION_STOPPED_DURING_IMPORT_329=Domain %s (server id: %s) : the sessi
WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328=Replication server RS(%d) not sending update \
%s for domain "%s" to server %d at %s because the replication protocol version %d \
negotiated with it has no encoding for this message
+ERR_INIT_REJECTED_SESSION_STOPPING_330=The total update of domain "%s" was refused by directory \
+ server %d: its session to the replication server is being stopped, and the entries would have \
+ been streamed over that session. Ask for the total update again once the session is back
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
index 23d4f018d6..538c7e1060 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -17,6 +17,7 @@
import static java.nio.charset.StandardCharsets.*;
import static org.assertj.core.api.Assertions.*;
+import static org.opends.messages.CoreMessages.ERR_UNCAUGHT_THREAD_EXCEPTION;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
@@ -26,7 +27,11 @@
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
@@ -40,6 +45,7 @@
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
+import org.opends.server.replication.protocol.ErrorMsg;
import org.opends.server.replication.protocol.InitializeRequestMsg;
import org.opends.server.replication.protocol.InitializeTargetMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
@@ -48,6 +54,7 @@
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
import org.testng.Assert;
@@ -75,6 +82,11 @@
* change is replayed while the import is waiting for them - or, for the request, while the
* exporter is holding the answer.
*
+ * The claim of a total update this replica did not ask for is made by the listener thread
+ * under no lock, so a restart of the session which reads no owner a moment before that claim
+ * would stop the session the import is about to read (issue #1041): the listener is held
+ * before its claim, and what stops the session is driven through the gap.
+ *
* The {@code timeOut} each case declares is what it is expected to take at the most; it is
* not what bounds it. {@code TestListener} sets the timeout of every test method from the
* {@code org.opends.test.timeout} property, ten minutes under Maven and none outside it.
@@ -508,6 +520,421 @@ public void aWarningAfterTheImportDoesNotCountTheDeliveriesBefore() throws Excep
}
}
+ /**
+ * A session restart decided after the {@code InitializeTargetMsg} was taken off the session
+ * and before the import claimed its context must not have the import run over the session
+ * it stops (issue #1041).
+ *
+ * The owner read of the restart and the claim of the listener share no lock: the restart
+ * reads no owner, stops the broker and waits for the listener thread to end - which is the
+ * thread about to run the import. Run over that broker, the import ends on the nothing
+ * which arrived - as a failed import since issue #1039, and as a finished one before it -
+ * over a suffix which has been replaced by it all the same. Here the listener is held
+ * before its claim, the restart is driven through the gap by a change whose attempts in
+ * place are spent and held between its decision and the stop, and the listener is released
+ * in between: the broker it finds is still up, so what refuses the import is the claim of
+ * the restart, and the refusal reaches the exporter over the session which is about to be
+ * stopped.
+ */
+ @Test(timeOut = 120_000)
+ public void aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final int totalUpdatesStartedBefore =
+ errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
+ final int totalUpdatesEndedBefore =
+ errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ final int listenerDeathsBefore = listenerDeaths().size();
+ final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size();
+
+ // The listener thread has taken the InitializeTargetMsg off the session and is held
+ // before it claims the import; the restart is held after its decision, before the stop.
+ final CountDownLatch listenerHeld = new CountDownLatch(1);
+ final CountDownLatch releaseListener = new CountDownLatch(1);
+ final CountDownLatch stopHeld = new CountDownLatch(1);
+ final CountDownLatch releaseStop = new CountDownLatch(1);
+ domain.setImportClaimHook(() -> {
+ listenerHeld.countDown();
+ awaitUninterruptibly(releaseListener);
+ });
+ domain.setServiceStopHook(() -> {
+ stopHeld.countDown();
+ awaitUninterruptibly(releaseStop);
+ });
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW));
+ assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
+ "the listener thread did not reach the claim of the import");
+
+ /*
+ * A change whose entryUUID search never runs spends its attempts in place, finds no
+ * owner and restarts the session. On a thread of its own: the restart is held before
+ * the stop, and then waits for the listener thread.
+ */
+ final CSN csn = gen.newCSN();
+ final AtomicReference
+ * Nothing claims against the listener here - the domain disabling itself stops the session
+ * whatever owns it - so what refuses the import is the listener reading, once its claim is
+ * made, that the broker it would stream over is stopping. Without that read the claim wins,
+ * and what runs next publishes the full update status over a session which is gone.
+ */
+ @Test(timeOut = 120_000)
+ public void aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=survivor," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: survivor",
+ "sn: survivor");
+ final int totalUpdatesStartedBefore =
+ errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
+ final int listenerDeathsBefore = listenerDeaths().size();
+ final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size();
+
+ final CountDownLatch listenerHeld = new CountDownLatch(1);
+ final CountDownLatch releaseListener = new CountDownLatch(1);
+ domain.setImportClaimHook(() -> {
+ listenerHeld.countDown();
+ awaitUninterruptibly(releaseListener);
+ });
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW));
+ assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
+ "the listener thread did not reach the claim of the import");
+
+ // On a thread of its own: disabling the domain waits for the listener thread.
+ final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN);
+ disable.start();
+ waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session");
+ releaseListener.countDown();
+ disable.join(60_000);
+ assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread"
+ + " it waits for is still there");
+ }
+ finally
+ {
+ releaseListener.countDown();
+ domain.setImportClaimHook(null);
+ }
+ domain.enable();
+ waitUntil(domain::isConnected, "the session was not started back by enable()");
+ assertFalse(domain.ieRunning(), "the refused import left its context claimed");
+
+ assertTrue(entryExists(entry.getName()), "the import ran over the session the disable"
+ + " stopped: the suffix was replaced by the nothing which arrived");
+ assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
+ .as("a total update claimed against a session which is being stopped was started")
+ .hasSize(totalUpdatesStartedBefore);
+ assertThat(listenerDeaths())
+ .as("the listener thread ended on an uncaught exception")
+ .hasSize(listenerDeathsBefore);
+ // Every record is written twice - the error log has two publishers in the tests.
+ assertThat(errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()))
+ .as("the refusal of the total update was not recorded on this server")
+ .hasSizeGreaterThan(refusalsBefore);
+ }
+
+ /**
+ * A domain disabled after the answer to a total update this replica asked for was taken off
+ * the session, and before the import started, must refuse the import too.
+ *
+ * The context is the one the request claimed, so there is nothing to claim against: what
+ * refuses the import is the same read of the broker as for a total update another server
+ * started. Without it the import runs over the session the disable stopped, and replaces the
+ * suffix with the nothing which arrived.
+ */
+ @Test(timeOut = 120_000)
+ public void aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=survivor," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: survivor",
+ "sn: survivor");
+ final int totalUpdatesStartedBefore =
+ errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
+ final int listenerDeathsBefore = listenerDeaths().size();
+
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CountDownLatch listenerHeld = new CountDownLatch(1);
+ final CountDownLatch releaseListener = new CountDownLatch(1);
+ domain.setImportClaimHook(() -> {
+ listenerHeld.countDown();
+ awaitUninterruptibly(releaseListener);
+ });
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, DS_ID, exportedEntries().length, INIT_WINDOW));
+ assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
+ "the listener thread did not reach the start of the import");
+
+ // On a thread of its own: disabling the domain waits for the listener thread.
+ final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN);
+ disable.start();
+ waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session");
+ releaseListener.countDown();
+ disable.join(60_000);
+ assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread"
+ + " it waits for is still there");
+ }
+ finally
+ {
+ releaseListener.countDown();
+ domain.setImportClaimHook(null);
+ }
+ domain.enable();
+ waitUntil(domain::isConnected, "the session was not started back by enable()");
+ assertFalse(domain.ieRunning(), "the refused import left the context of its request claimed");
+
+ assertTrue(entryExists(entry.getName()), "the import ran over the session the disable"
+ + " stopped: the suffix was replaced by the nothing which arrived");
+ assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
+ .as("a total update answered over a session which is being stopped was started")
+ .hasSize(totalUpdatesStartedBefore);
+ assertThat(listenerDeaths())
+ .as("the listener thread ended on an uncaught exception")
+ .hasSize(listenerDeathsBefore);
+ }
+
+ /**
+ * A session restart decided while a total update out of this replica is running stops the
+ * session that export streams over (issue #1041).
+ *
+ * What the restart must leave alone is a total update into this replica: the data it is
+ * about to replace is read over the session, and the import is the thread the stop waits
+ * for. An export is not that: it streams out of a backend nothing is taking away, on a
+ * thread of its own, and a session stopped under it is the cut it reports to whoever asked
+ * for the total update - the same cut every other stop of the session is. The claim the
+ * restart makes for the import is not made here, and the session is stopped as it was
+ * before the claim.
+ *
+ * The export holds the context by standing where it waits for its target to report the
+ * start of the total update: the target is a broker of this test, and reports nothing.
+ */
+ @Test(timeOut = 120_000)
+ public void aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ waitUntil(() -> domain.getReplicaInfos().containsKey(EXPORTER_ID),
+ "the exporter is not in the replicas of the domain: nothing to export into");
+
+ final AtomicReference