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/main/java/org/opends/server/replication/service/ReplicationDomain.java | 201 ++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 179 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;
--
Gitblit v1.10.0