From c6650aedac5c1ade8aee9e7a76eb0dd100a3e179 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 02 Nov 2026 15:39:56 +0000
Subject: [PATCH] [#1246] Keep the messages of the last local changes, and have a recovery publish the changes done before it began from them, as they were done (#1251)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java | 81 +++++++++++++++++++++++++++++++++++-----
1 files changed, 71 insertions(+), 10 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
index fc2fb21..ee63266 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
@@ -13,7 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2023-2026 3A Systems LLC.
+ * Portions Copyright 2023-2026 3A Systems, LLC.
* Portions Copyright 2025 Wren Security.
*/
package org.opends.server.replication.service;
@@ -236,6 +236,8 @@
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
private volatile boolean shutdown;
+ /** A backend whose state could not be restored must never reconnect through this broker. */
+ private volatile boolean permanentlyStopped;
/**
* The stop in progress: set by {@link #stop()} before anything else it does, and cleared to
* {@code null} by {@link #start()}. While it is set, a publish which finds no window credit by
@@ -366,7 +368,7 @@
{
synchronized (startStopLock)
{
- if (!shutdown)
+ if (!shutdown || permanentlyStopped)
{
return;
}
@@ -2300,7 +2302,7 @@
// Synchronize inside the loop in order to allow shutdown.
synchronized (startStopLock)
{
- if (rs.isConnected() || shutdown)
+ if (rs.isConnected() || shutdown || permanentlyStopped)
{
break;
}
@@ -2416,7 +2418,7 @@
{
boolean done = false;
- while (!done && !shutdown && !gaveUpForTheStop())
+ while (!done && !shutdown && !permanentlyStopped && !gaveUpForTheStop())
{
if (connectionError)
{
@@ -2508,7 +2510,7 @@
Check the session. If it has changed, some disconnection or
reconnection happened and we need to restart from scratch.
*/
- if (gaveUpForTheStop())
+ if (permanentlyStopped || gaveUpForTheStop())
{
/*
* The stop gave up since the loop head, while this publish waited for this lock - for
@@ -2559,7 +2561,7 @@
Not once the stop gave up, which it may have done while this waited for the lock.
*/
Session session = connectedRS.get().session;
- if (session != null && !gaveUpForTheStop())
+ if (session != null && !permanentlyStopped && !gaveUpForTheStop())
{
session.publish(new WindowProbeMsg());
}
@@ -2659,7 +2661,7 @@
boolean reconnectOnFailure, boolean returnOnTopoChange)
throws SocketTimeoutException
{
- while (!shutdown)
+ while (!shutdown && !permanentlyStopped)
{
ConnectedRS rs = connectedRS.get();
if (!rs.isConnected())
@@ -2896,9 +2898,48 @@
}
}
+ /**
+ * Stops a domain with untrusted data state. No offline CSN is announced and buffered messages
+ * are discarded rather than drained to the peer. A later configuration change cannot restart it.
+ */
+ void stopPermanently()
+ {
+ permanentlyStopped = true;
+ final Session session = connectedRS.get().session;
+ if (session != null)
+ {
+ session.abort(new IOException("The total update could not restore its backend state"));
+ }
+ stop();
+ }
+
+ /** Returns whether this broker has been permanently stopped following an import failure. */
+ boolean isPermanentlyStopped()
+ {
+ return permanentlyStopped;
+ }
+
/** Stop the server. */
public void stop()
{
+ stop(true);
+ }
+
+ /**
+ * Stops an aborted import on its listener without waiting for credit that only that listener
+ * can receive. The next enable may start a new session; no offline CSN is announced.
+ * An ECL-enabled replica therefore leaves its last known CSN in the replication server's
+ * consistency point, as on an abrupt disconnect: new change-number assignment can stall across
+ * its ECL-enabled domains until this replica reconnects and advances that CSN. Having send credit
+ * alone would not make an offline announcement safe here: its locks or socket write can still block.
+ */
+ void stopAfterAbortedImport()
+ {
+ stop(false);
+ }
+
+ private void stop(boolean announceOffline)
+ {
if (logger.isTraceEnabled() && !shutdown)
{
debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
@@ -2921,13 +2962,33 @@
try
{
// In the try: a watchdog which cannot be started must not leave the broker stopping either.
- if (newStop && stop.bounded)
+ if (newStop && stop.bounded && announceOffline)
{
new DirectoryThread(() -> endWhatStillHoldsTheStop(stop),
"Replica DS(" + getServerId() + ") watchdog of the stop for domain \"" + getBaseDN() + "\"")
.start();
}
- domain.publishReplicaOfflineMsg();
+ if (!announceOffline)
+ {
+ synchronized (stop)
+ {
+ if (!stop.ended)
+ {
+ // End blocked writes too: closing in order could wait for the peer to read forever.
+ // stop.end() takes this monitor, so this close cannot reach a later session.
+ stop.gaveUp = true;
+ final Session session = connectedRS.get().session;
+ if (session != null)
+ {
+ session.abort(new IOException("The total update preparation left its backend unavailable"));
+ }
+ }
+ }
+ }
+ else if (!permanentlyStopped)
+ {
+ domain.publishReplicaOfflineMsg();
+ }
}
finally
{
@@ -3674,7 +3735,7 @@
*/
boolean shuttingDown()
{
- return shutdown;
+ return shutdown || permanentlyStopped;
}
/**
--
Gitblit v1.10.0