[#1246] Keep the messages of the last local changes, and have a recovery pu...

Valery Kharseko

15 hours ago c6650aedac5c1ade8aee9e7a76eb0dd100a3e179

 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;
  }

  /**