From 641ff2ce8441a33717bdf066c3b0ae2fe96ba271 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 12 Aug 2026 18:20:32 +0000
Subject: [PATCH] [#861] Fail fast when a total update request gets no answer (#864)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 150 +++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java | 9 +
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 201 +++++++++++++++++++++++++++++---
3 files changed, 338 insertions(+), 22 deletions(-)
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 cc05d77..dcc7dff 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
@@ -13,8 +13,7 @@
*
* Copyright 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2025-2026 3A Systems LLC.
- * Portions Copyright 2026 3A Systems, LLC.
+ * Portions Copyright 2025-2026 3A Systems, LLC.
*/
package org.opends.server.replication.service;
@@ -821,9 +820,12 @@
" Error Msg received: " + errorMsg);
}
- if (errorMsg.getCreationTime() > ieCtx.startTime)
+ // consider only ErrorMsg that relate to the current import/export.
+ // ">=" and not ">": with all servers on one host the whole
+ // request/rejection round-trip can complete within the millisecond
+ // this context was created in (issue #861)
+ if (errorMsg.getCreationTime() >= ieCtx.startTime)
{
- // consider only ErrorMsg that relate to the current import/export
processErrorMsg(errorMsg, ieCtx);
}
else
@@ -1146,11 +1148,28 @@
private InitializeRequestMsg initReqMsgSent;
/**
- * Start time of the initialization process. ErrorMsg timestamped before
- * this startTime will be ignored.
+ * Start time of the initialization process. ErrorMsg timestamped strictly
+ * before this startTime will be ignored.
*/
private final long startTime;
+ /**
+ * Time when {@link #initReqMsgSent} was last published. Volatile: written
+ * by the requesting and listener threads, read by the task thread running
+ * {@link ReplicationDomain#abortStalledInitializeFromRemote(long)}.
+ */
+ private volatile long requestSentTime;
+ /**
+ * Whether the InitializeTargetMsg answering {@link #initReqMsgSent} has
+ * been received. Guarded by this context's monitor.
+ */
+ private boolean startReceived;
+ /**
+ * Whether the stalled-request watchdog abandoned this context. Guarded by
+ * this context's monitor.
+ */
+ private boolean abandonedAsStalled;
+
/** List for replicas (DS) connected to the topology when initialization started. */
private final Set<Integer> startList = new HashSet<>(0);
@@ -1202,6 +1221,58 @@
}
/**
+ * Returns the start time of this initialization, for tests.
+ *
+ * @return the creation time of this context in milliseconds
+ */
+ long getStartTime()
+ {
+ return startTime;
+ }
+
+ /** Arms (or re-arms, on a new attempt) the stalled-request watchdog. */
+ private synchronized void markInitRequestSent()
+ {
+ startReceived = false;
+ requestSentTime = System.currentTimeMillis();
+ }
+
+ /**
+ * Marks that the InitializeTargetMsg answering the published
+ * InitializeRequestMsg has been received.
+ *
+ * @return false when the stalled-request watchdog already abandoned this
+ * context, in which case the start message must be ignored
+ */
+ private synchronized boolean markInitStartReceived()
+ {
+ if (abandonedAsStalled)
+ {
+ return false;
+ }
+ startReceived = true;
+ return true;
+ }
+
+ /**
+ * Abandons this context when the published request has received no answer
+ * within the provided delay.
+ *
+ * @param stalledTimeoutMs delay after which the request is considered lost
+ * @return whether this call abandoned the context
+ */
+ private synchronized boolean abandonIfStalled(long stalledTimeoutMs)
+ {
+ if (startReceived || abandonedAsStalled
+ || System.currentTimeMillis() - requestSentTime < stalledTimeoutMs)
+ {
+ return false;
+ }
+ abandonedAsStalled = true;
+ return true;
+ }
+
+ /**
* Returns the total number of entries to be processed when a total update
* is in progress.
*
@@ -1326,7 +1397,10 @@
*/
public void setExceptionIfNoneSet(DirectoryException exception)
{
- if (exception == null)
+ // Historical upstream bug (since at least OpenDJ 3): the null check was
+ // made on the argument instead of the field, so no error was ever
+ // recorded and every failed total update completed "successfully"
+ if (this.exception == null)
{
this.exception = exception;
}
@@ -2064,7 +2138,7 @@
if (ieCtx.getException() == null)
{
ErrorMsg errMsg = (ErrorMsg)msg;
- if (errMsg.getCreationTime() > ieCtx.startTime)
+ if (errMsg.getCreationTime() >= ieCtx.startTime)
{
ieCtx.setException(
new DirectoryException(ResultCode.OTHER,errMsg.getDetails()));
@@ -2308,7 +2382,16 @@
ieCtx.attemptCnt = 0;
ieCtx.initReqMsgSent = new InitializeRequestMsg(
getBaseDN(), getServerId(), source, getInitWindow());
- broker.publish(ieCtx.initReqMsgSent);
+ ieCtx.markInitRequestSent();
+ // The broker silently drops the message when it is caught between two
+ // sessions (connection error, recovery pending after a reconnect) and
+ // only replays UpdateMsgs on reconnect: an unpublished request would
+ // leave the task waiting forever for an answer (issue #861).
+ if (!broker.publish(ieCtx.initReqMsgSent, true))
+ {
+ throw new DirectoryException(ResultCode.OTHER,
+ ERR_INITIALIZATION_FAILED_NOCONN.get(getBaseDN()));
+ }
/*
The normal success processing is now to receive InitTargetMsg then
@@ -2340,6 +2423,42 @@
}
/**
+ * Fails the on-going initialization from a remote replica when the request
+ * published by {@link #initializeFromRemote(int, Task)} has received no
+ * answer at all - neither the InitializeTargetMsg starting the import nor an
+ * ErrorMsg - within the provided delay. The request or its answer can be
+ * lost with no error ever coming back (issue #861), and nothing else bounds
+ * the wait: without this watchdog the initialize task hangs forever.
+ *
+ * @param stalledTimeoutMs
+ * delay in milliseconds after which the unanswered request is
+ * considered lost
+ * @return whether a stalled initialization was aborted by this call
+ */
+ public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs)
+ {
+ final ImportExportContext ieCtx = importExportContext.get();
+ if (ieCtx == null || !ieCtx.importInProgress() || ieCtx.initReqMsgSent == null
+ || !ieCtx.abandonIfStalled(stalledTimeoutMs))
+ {
+ return false;
+ }
+ // Once abandonIfStalled() returned true a concurrently received
+ // InitializeTargetMsg is ignored by the listener, so releasing the
+ // context here cannot race the start of an import.
+ ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
+ ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(
+ getBaseDN(), ieCtx.initReqMsgSent.getDestination())));
+ if (ieCtx.initializeTask instanceof InitializeTask)
+ {
+ ((InitializeTask) ieCtx.initializeTask)
+ .updateTaskCompletionState(ieCtx.getException());
+ }
+ releaseIEContext();
+ return true;
+ }
+
+ /**
* Processes an InitializeTargetMsg received from a remote server
* meaning processes an initialization from the entries expected to be
* received now.
@@ -2359,8 +2478,48 @@
}
InitializeTask initFromTask = null;
- int source = initTargetMsgReceived.getSenderID();
- ImportExportContext ieCtx = importExportContext.get();
+ final int source = initTargetMsgReceived.getSenderID();
+ final ImportExportContext ieCtx;
+ if (initTargetMsgReceived.getInitiatorID() == getServerId())
+ {
+ ieCtx = importExportContext.get();
+ if (ieCtx == null || !ieCtx.markInitStartReceived())
+ {
+ /*
+ The stalled-request watchdog abandoned the initialization this message
+ answers (issue #861): its task already failed and its context is (about
+ to be) released. The entries following this message are discarded by
+ the listener until the exporter completes.
+ */
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("[IE] Ignoring InitializeTargetMsg from server " + source
+ + " for domain " + getBaseDN()
+ + ": the initialization was abandoned as stalled");
+ }
+ return;
+ }
+ }
+ else
+ {
+ /*
+ The initTargetMsgReceived is for an import initiated by the remote
+ server. Test and set if no import already in progress
+ */
+ try
+ {
+ ieCtx = acquireIEContext(true);
+ }
+ 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()));
+ return;
+ }
+ }
try
{
// Log starting
@@ -2370,16 +2529,6 @@
// Go into full update status
setNewStatus(StatusMachineEvent.TO_FULL_UPDATE_STATUS_EVENT);
- // Acquire an import context if no already done (and initialize).
- if (initTargetMsgReceived.getInitiatorID() != getServerId())
- {
- /*
- The initTargetMsgReceived is for an import initiated by the remote server.
- Test and set if no import already in progress
- */
- ieCtx = acquireIEContext(true);
- }
-
// Initialize stuff
ieCtx.importSource = source;
ieCtx.initializeCounters(initTargetMsgReceived.getEntryCount());
@@ -2435,7 +2584,15 @@
logger.info(NOTE_RESENDING_INIT_FROM_REMOTE_REQUEST,
ieCtx.getException().getLocalizedMessage());
- broker.publish(ieCtx.initReqMsgSent);
+ ieCtx.markInitRequestSent();
+ if (!broker.publish(ieCtx.initReqMsgSent, true))
+ {
+ // Same silent-drop hazard as the first request (issue #861):
+ // fail the attempt instead of waiting for an answer that
+ // cannot arrive
+ throw new DirectoryException(ResultCode.OTHER,
+ ERR_INITIALIZATION_FAILED_NOCONN.get(getBaseDN()));
+ }
ieCtx.initializeCounters(0);
ieCtx.exception = null;
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java b/opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java
index d3c080d..f9eb7c3 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java
@@ -40,6 +40,14 @@
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /**
+ * Delay after which the initialization is considered stalled when the
+ * request for entries has received no answer from the topology: the request
+ * or its response can be silently lost with no error reported back (issue
+ * #861), and no other mechanism bounds this wait.
+ */
+ private static final long INITIALIZE_START_TIMEOUT_MS = 2 * 60 * 1000L;
+
private String domainString;
private int source;
private LDAPReplicationDomain domain;
@@ -120,6 +128,7 @@
initStateLock.wait(1000);
replaceAttributeValue(ATTR_TASK_INITIALIZE_LEFT, String.valueOf(left));
replaceAttributeValue(ATTR_TASK_INITIALIZE_DONE, String.valueOf(total-left));
+ domain.abortStalledInitializeFromRemote(INITIALIZE_START_TIMEOUT_MS);
}
}
replaceAttributeValue(ATTR_TASK_INITIALIZE_LEFT, String.valueOf(left));
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
index fd03a26..662e1a5 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
@@ -27,6 +27,7 @@
import java.util.Map;
import java.util.SortedSet;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -47,6 +48,7 @@
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationDomain.ImportExportContext;
import org.forgerock.opendj.ldap.DN;
+import org.opends.server.tasks.InitializeTask;
import org.opends.server.types.DirectoryException;
import org.opends.server.util.TestTimer;
import org.testng.annotations.DataProvider;
@@ -608,6 +610,154 @@
}
}
+ /** InitializeTask double recording the completion reported by the domain. */
+ private static final class RecordingInitializeTask extends InitializeTask
+ {
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private volatile DirectoryException failure;
+
+ @Override
+ public void updateTaskCompletionState(DirectoryException de)
+ {
+ failure = de;
+ completed.countDown();
+ }
+
+ DirectoryException waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException
+ {
+ assertTrue(completed.await(timeout, unit), "the initialize task never completed");
+ return failure;
+ }
+ }
+
+ /**
+ * An ErrorMsg answering an initialization request can be created within the
+ * same millisecond as the requester's import/export context when the whole
+ * topology runs on one host: it must terminate the pending initialization
+ * instead of being discarded as stale (issue #861).
+ */
+ @Test(enabled=true)
+ public void errorMsgFromSameMillisecondTerminatesPendingInitialize() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer = null;
+ FakeReplicationDomain domain2 = null;
+ ReplicationBroker broker3 = null;
+
+ try
+ {
+ int replServerPort = TestCaseUtils.findFreePort();
+ replServer = createReplicationServer(13, replServerPort,
+ "sameMillisecondErrorMsgDb", 100);
+ SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);
+
+ domain2 = new FakeReplicationDomain(
+ testService, 2, servers, 0, null, new StringBuffer(), 0);
+ broker3 = openReplicationSession(testService, 3, 100, replServerPort,
+ 10000, domain2.getGenerationID());
+
+ waitTopologyKnowsReplica(domain2, 3);
+
+ RecordingInitializeTask task = new RecordingInitializeTask();
+ domain2.initializeFromRemote(3, task);
+ long startTime = domain2.getImportExportContext().getStartTime();
+
+ // strictly older than the context: still ignored as stale
+ ErrorMsg staleError = new ErrorMsg(3, 2, LocalizableMessage.raw("stale error"));
+ staleError.setCreationTime(startTime - 1);
+ broker3.publish(staleError);
+
+ // same millisecond as the context: must terminate the initialization
+ ErrorMsg currentError = new ErrorMsg(3, 2, LocalizableMessage.raw("current error"));
+ currentError.setCreationTime(startTime);
+ broker3.publish(currentError);
+
+ DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS);
+ assertNotNull(failure, "the initialization completed without an error");
+ assertEquals(failure.getMessageObject().toString(), "current error",
+ "the ErrorMsg timestamped before the context must stay ignored");
+ assertFalse(domain2.ieRunning(),
+ "the terminated initialization must release the import/export context");
+ }
+ finally
+ {
+ stop(broker3);
+ disable(domain2);
+ remove(replServer);
+ }
+ }
+
+ /**
+ * When the initialization request receives no answer at all - the publish
+ * was silently dropped or the answer was lost - the stalled-request watchdog
+ * must fail the task after the configured delay instead of letting it wait
+ * forever (issue #861).
+ */
+ @Test(enabled=true)
+ public void stalledInitializeFromRemoteIsAborted() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer = null;
+ FakeReplicationDomain domain2 = null;
+ ReplicationBroker broker3 = null;
+
+ try
+ {
+ int replServerPort = TestCaseUtils.findFreePort();
+ replServer = createReplicationServer(14, replServerPort,
+ "stalledInitializeRequestDb", 100);
+ SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);
+
+ domain2 = new FakeReplicationDomain(
+ testService, 2, servers, 0, null, new StringBuffer(), 0);
+ // broker3 receives the InitializeRequestMsg and never answers it
+ broker3 = openReplicationSession(testService, 3, 100, replServerPort,
+ 10000, domain2.getGenerationID());
+
+ waitTopologyKnowsReplica(domain2, 3);
+
+ RecordingInitializeTask task = new RecordingInitializeTask();
+ domain2.initializeFromRemote(3, task);
+
+ assertFalse(domain2.abortStalledInitializeFromRemote(60000),
+ "the initialization must not be aborted before the delay elapses");
+
+ final FakeReplicationDomain requester = domain2;
+ TestTimer abortTimer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(10, MILLISECONDS)
+ .toTimer();
+ abortTimer.repeatUntilSuccess(() -> assertTrue(
+ requester.abortStalledInitializeFromRemote(50),
+ "the stalled initialization was never aborted"));
+
+ DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS);
+ assertNotNull(failure, "the stalled initialization must fail the task");
+ assertEquals(failure.getMessageObject().toString(),
+ ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(testService, 3).toString());
+ assertFalse(domain2.ieRunning(),
+ "the aborted initialization must release the import/export context");
+ assertFalse(domain2.abortStalledInitializeFromRemote(0),
+ "a second abort must be a no-op once the context is released");
+ }
+ finally
+ {
+ stop(broker3);
+ disable(domain2);
+ remove(replServer);
+ }
+ }
+
+ private void waitTopologyKnowsReplica(ReplicationDomain domain, int dsId) throws Exception
+ {
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(() -> assertTrue(domain.getReplicaInfos().containsKey(dsId),
+ "DS(" + dsId + ") is not known to the domain"));
+ }
+
private String buildExportedData(final int ENTRYCOUNT)
{
final StringBuilder sb = new StringBuilder();
--
Gitblit v1.10.0