[#1207] Route total updates to a replica that has the server ID of a replic...

Valery Kharseko

19 hours ago 21446c9656ae9cde8f67d1ddb66cd2f99e9a23fb

 opendj-server-legacy/src/main/java/org/opends/admin/ads/ReplicaDescriptor.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2007-2009 Sun Microsystems, Inc.
 * Portions Copyright 2013-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.admin.ads;

@@ -33,7 +34,7 @@
  private int nbEntries = -1;
  private ServerDescriptor server;
  private final Set<HostPort> replicationServers = new HashSet<>();
  /** @see InstallerHelper#getReplicationId(Set) */
  /** @see InstallerHelper#addNewReplicationId(Set) */
  private int serverId = -1;
  private int missingChanges = -1;
  private long ageOfOldestMissingChange = -1;

 opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/Installer.java

@@ -2166,41 +2166,23 @@
    notifyListeners(getFormattedWithPoints(INFO_PROGRESS_CONFIGURING_REPLICATION.get()));

    InstallerHelper helper = new InstallerHelper();
    Set<Integer> knownReplicaServerIds = new HashSet<>();
    Set<Integer> knownReplicationServerIds = new HashSet<>();
    final Set<Integer> knownServerIds;
    if (lastLoadedCache != null)
    {
      for (SuffixDescriptor suffix : lastLoadedCache.getSuffixes())
      {
        for (ReplicaDescriptor replica : suffix.getReplicas())
        {
          knownReplicaServerIds.add(replica.getServerId());
        }
      }
      for (ServerDescriptor server : lastLoadedCache.getServers())
      {
        Object v = server.getServerProperties().get(REPLICATION_SERVER_ID);
        if (v != null)
        {
          knownReplicationServerIds.add((Integer) v);
        }
      }
      knownServerIds = InstallerHelper.getUsedServerIds(lastLoadedCache.getServers());
    }
    else
    {
      /* There is no ADS anywhere. Just use the SuffixDescriptors we found */
      Set<ServerDescriptor> servers = new HashSet<>();
      for (SuffixDescriptor suffix : getUserData().getSuffixesToReplicateOptions().getAvailableSuffixes())
      {
        for (ReplicaDescriptor replica : suffix.getReplicas())
        {
          knownReplicaServerIds.add(replica.getServerId());
          Object v = replica.getServer().getServerProperties().get(REPLICATION_SERVER_ID);
          if (v != null)
          {
            knownReplicationServerIds.add((Integer) v);
          }
          servers.add(replica.getServer());
        }
      }
      knownServerIds = InstallerHelper.getUsedServerIds(servers);
    }

    /*
@@ -2258,8 +2240,7 @@
      helper.configureReplication(conn, replicationServers,
          getUserData().getReplicationOptions().getReplicationPort(),
          getUserData().getReplicationOptions().useSecureReplication(),
          knownReplicationServerIds,
          knownReplicaServerIds);
          knownServerIds);
      localTimeMeasureTime = System.currentTimeMillis();
      localTime = Utils.getServerClock(conn);
      localServerDisplay = conn.getHostPort();
@@ -2353,8 +2334,7 @@
        try (ConnectionWrapper conn = getRemoteConnection(server))
        {
          ConfiguredReplication repl = helper.configureReplication(
              conn, remoteReplicationServers, replicationPort, enableSecureReplication,
              knownReplicationServerIds, knownReplicaServerIds);
              conn, remoteReplicationServers, replicationPort, enableSecureReplication, knownServerIds);
          long remoteTimeMeasureTime = System.currentTimeMillis();
          long remoteTime = Utils.getServerClock(conn);
          if (localTime != -1

 opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/InstallerHelper.java

@@ -69,6 +69,8 @@
import org.forgerock.opendj.server.config.meta.ReplicationServerCfgDefn;
import org.forgerock.opendj.server.config.meta.ReplicationSynchronizationProviderCfgDefn;
import org.forgerock.opendj.server.config.server.BackendCfg;
import org.opends.admin.ads.ReplicaDescriptor;
import org.opends.admin.ads.ServerDescriptor;
import org.opends.admin.ads.util.ConnectionWrapper;
import org.opends.guitools.controlpanel.util.Utilities;
import org.opends.messages.BackendMessages;
@@ -388,17 +390,17 @@
   * configured (it might not exist and the user specified it in the setup).
   * @param useSecureReplication whether to encrypt connections with the
   * replication port or not.
   * @param usedReplicationServerIds the list of replication server ids that
   * are already used.
   * @param usedReplicaServerIds the list of server ids (domain ids) that
   * are already used by replicas.
   * @param usedServerIds the server ids that the replication servers and the
   * replication domains of the topology already use, in one set (see
   * {@link #getUsedServerIds(Collection)}). A new replication server or domain
   * takes its id outside of it, and the set is updated with the ids of the
   * replication server and of the domains of this server.
   * @throws ApplicationException if something goes wrong.
   * @return a ConfiguredReplication object describing what has been configured.
   */
  public ConfiguredReplication configureReplication(
      ConnectionWrapper conn, Map<DN, Set<HostPort>> replicationServers,
      int replicationPort, boolean useSecureReplication, Set<Integer> usedReplicationServerIds,
      Set<Integer> usedReplicaServerIds)
      int replicationPort, boolean useSecureReplication, Set<Integer> usedServerIds)
  throws ApplicationException
  {
    boolean synchProviderCreated;
@@ -477,8 +479,7 @@
        {
          secureReplicationEnabled = false;
        }
        int id = getReplicationId(usedReplicationServerIds);
        usedReplicationServerIds.add(id);
        int id = addNewReplicationId(usedServerIds);
        replicationServer = sync.createReplicationServer(
            ReplicationServerCfgDefn.getInstance(),
            new ArrayList<PropertyException>());
@@ -490,8 +491,7 @@
      {
        secureReplicationEnabled = false;
        replicationServer = sync.getReplicationServer();
        usedReplicationServerIds.add(
            replicationServer.getReplicationServerId());
        usedServerIds.add(replicationServer.getReplicationServerId());
        replicationServerCreated = false;
      }

@@ -542,8 +542,7 @@
        }
        if (domain == null)
        {
          int replicaServerId = getReplicationId(usedReplicaServerIds);
          usedReplicaServerIds.add(replicaServerId);
          int replicaServerId = addNewReplicationId(usedServerIds);
          domainName = getDomainName(domainNames, dn);
          domain = sync.createReplicationDomain(
              ReplicationDomainCfgDefn.getInstance(), domainName,
@@ -563,7 +562,7 @@
        }
        servers = toLowerCaseStrings(entry.getValue());
        domain.setReplicationServer(servers);
        usedReplicaServerIds.add(domain.getServerId());
        usedServerIds.add(domain.getServerId());

        domain.commit();
        Set<String> addedServers = intersect(servers, oldServers);
@@ -780,11 +779,48 @@
  }

  /**
   * Returns the ID to be used for a new replication server or domain.
   * @param usedIds the list of already used ids.
   * Returns the server IDs that the replication servers and the replication domains of the provided
   * servers use, in one set.
   * <p>
   * A new replication server and a new replication domain both take their ID outside of this set, so
   * that no replication domain of any base DN has the ID of a replication server: a replication
   * server of an earlier version takes the total update messages sent to its own ID as messages for
   * itself, and the assured replication counts the acknowledgments of a replication server and of a
   * directory server with the same ID as one. The set only holds what the descriptors carry: the
   * domains of the backend base DNs they were loaded for, not a domain of another base DN such as one
   * below a backend base DN, and nothing of a server that has no descriptor.
   *
   * @param servers the servers of the topology.
   * @return the server IDs that the replication servers and the replication domains of the provided
   *         servers use.
   */
  public static Set<Integer> getUsedServerIds(Collection<ServerDescriptor> servers)
  {
    final Set<Integer> ids = new HashSet<>();
    for (ServerDescriptor server : servers)
    {
      if (server.isReplicationServer())
      {
        ids.add(server.getReplicationServerId());
      }
      for (ReplicaDescriptor replica : server.getReplicas())
      {
        if (replica.isReplicated())
        {
          ids.add(replica.getServerId());
        }
      }
    }
    return ids;
  }

  /**
   * Returns the ID to be used for a new replication server or domain, and adds it to the used ids,
   * so that the next new replication server or domain takes another one (issue #1207).
   * @param usedIds the already used ids. The new ID is taken outside of them and added to them.
   * @return the ID to be used for a new replication server or domain.
   */
  public static int getReplicationId(Set<Integer> usedIds)
  public static int addNewReplicationId(Set<Integer> usedIds)
  {
    Random r = new Random();
    int id = 0;
@@ -792,6 +828,7 @@
    {
      id = r.nextInt(MAX_ID_VALUE);
    }
    usedIds.add(id);
    return id;
  }


 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -13,7 +13,7 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

@@ -150,6 +150,19 @@
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /**
   * The id of the message of the warning a peer replication server sends when the generation ids
   * differ, in the form an ErrorMsg carries it on the wire since OpenDJ 3.0.
   */
  private static final String GENERATION_ID_WARNING_MSG_ID =
      WARN_BAD_GENERATION_ID_FROM_RS.resourceName() + "-" + WARN_BAD_GENERATION_ID_FROM_RS.ordinal();

  /**
   * The id an OpenDJ 2.6 replication server gives the same warning: the numeric id of its message
   * descriptor, category SYNC (0x00E00000), severity SEVERE_WARNING (0x00020000), ordinal 77.
   */
  private static final String GENERATION_ID_WARNING_MSG_ID_2X = String.valueOf(0x00E00000 | 0x00020000 | 77);

  /**
   * The needed info for each received assured update message we are waiting
   * acks for.
   * <p>
@@ -1194,13 +1207,13 @@
            stopMonitoringPublisher();
          }

          if (connectedRSs.containsKey(sHandler.getServerId()))
          // The handler, not its server id, tells which of the two maps holds it: a directory server
          // can have the server id of a replication server (issue #1207)
          final boolean isDataServer = !sHandler.isReplicationServer();
          final Map<Integer, ? extends ServerHandler> connectedServers = isDataServer ? connectedDSs : connectedRSs;
          if (connectedServers.containsKey(sHandler.getServerId()))
          {
            unregisterServerHandler(sHandler, shutdown, false);
          }
          else if (connectedDSs.containsKey(sHandler.getServerId()))
          {
            unregisterServerHandler(sHandler, shutdown, true);
            unregisterServerHandler(sHandler, shutdown, isDataServer);
          }
        }
        catch(Exception e)
@@ -1569,6 +1582,12 @@
  /**
   * Processes a message coming from one server in the topology and potentially
   * forwards it to one or all other servers.
   * <p>
   * A directory server that has the server id of this replication server gets the messages sent to
   * that id: a replication server holds no data to export or import, so the total update messages
   * routed by id are for the directory server. Only the warning of a peer replication server about
   * the generation id, and a message no directory server with the id can get, are for this
   * replication server.
   *
   * @param msg
   *          The message received and to be processed.
@@ -1577,7 +1596,18 @@
   */
  void process(RoutableMsg msg, ServerHandler sender)
  {
    if (msg.getDestination() == localReplicationServer.getServerId())
    if (isGenerationIdWarningOfAPeer(msg, sender))
    {
      logger.error(ERR_ERROR_MSG_RECEIVED, ((ErrorMsg) msg).getDetails());
      return;
    }

    List<ServerHandler> servers = getDestinationServers(msg, sender);
    if (!servers.isEmpty())
    {
      forwardMsgToAllServers(msg, servers, sender);
    }
    else if (msg.getDestination() == localReplicationServer.getServerId())
    {
      // Handle routable messages targeted at this RS.
      if (msg instanceof ErrorMsg)
@@ -1592,20 +1622,36 @@
    }
    else
    {
      // Forward message not destined for this RS.
      List<ServerHandler> servers = getDestinationServers(msg, sender);
      if (!servers.isEmpty())
      {
        forwardMsgToAllServers(msg, servers, sender);
      }
      else
      {
        replyWithUnreachablePeerMsg(sender, msg);
      }
      replyWithUnreachablePeerMsg(sender, msg);
    }
  }

  /**
   * Returns whether the message is the warning a peer replication server sends this one when their
   * generation ids differ (see {@link #receiveTopoInfoFromRS}). A directory server with the server id
   * of this replication server must not get it: it would take it for the end of the total update it
   * is in. The warning is told from the other messages a peer replication server sends to that id by
   * the id of its message (see {@link #isGenerationIdWarningMsgID(String)}).
   */
  private boolean isGenerationIdWarningOfAPeer(RoutableMsg msg, ServerHandler sender)
  {
    return sender.isReplicationServer()
        && msg.getDestination() == localReplicationServer.getServerId()
        && msg instanceof ErrorMsg
        && isGenerationIdWarningMsgID(((ErrorMsg) msg).getMsgID());
  }

  /**
   * Returns whether the id of the message an ErrorMsg carries is the one of the warning about the
   * generation id: {@code <resource name>-<ordinal>} as replication servers send it since OpenDJ 3.0,
   * or the numeric id of the message descriptor as an OpenDJ 2.6 replication server sends it.
   */
  static boolean isGenerationIdWarningMsgID(String msgID)
  {
    return GENERATION_ID_WARNING_MSG_ID.equals(msgID) || GENERATION_ID_WARNING_MSG_ID_2X.equals(msgID);
  }

  /**
   * Responds to a monitor request message.
   *
   * @param msg
@@ -2725,8 +2771,14 @@
  {
    startMonitoringPublisher();

    final int serverId = dsHandler.getServerId();
    if (serverId == localReplicationServer.getServerId() || connectedRSs.containsKey(serverId))
    {
      warnReplicaHasTheServerIdOfAReplicationServer(serverId);
    }

    // connected with new DS: store handler.
    connectedDSs.put(dsHandler.getServerId(), dsHandler);
    connectedDSs.put(serverId, dsHandler);

    // Tell peer RSs and DSs a new DS just connected to us
    // No need to re-send TopologyMsg to this just new DS
@@ -2743,9 +2795,26 @@
  {
    startMonitoringPublisher();

    final int serverId = rsHandler.getServerId();
    if (connectedDSs.containsKey(serverId))
    {
      warnReplicaHasTheServerIdOfAReplicationServer(serverId);
    }

    // connected with new RS (either outgoing or incoming
    // connection): store handler.
    connectedRSs.put(rsHandler.getServerId(), rsHandler);
    connectedRSs.put(serverId, rsHandler);
  }

  /**
   * Reports a directory server connected to this replication server that has the server id of a
   * replication server. This replication server routes the messages sent to that id to the
   * directory server, but a replication server of an earlier version keeps those messages for itself.
   */
  private void warnReplicaHasTheServerIdOfAReplicationServer(int serverId)
  {
    logger.warn(WARN_REPLICA_HAS_THE_SERVER_ID_OF_A_REPLICATION_SERVER,
        localReplicationServer.getMonitorInstanceName(), serverId, baseDN, serverId);
  }

  private void debug(String message)

 opendj-server-legacy/src/main/java/org/opends/server/tools/dsreplication/ReplicationCliMain.java

@@ -4593,11 +4593,10 @@
  private void updateConfiguration(ConnectionWrapper conn1, ConnectionWrapper conn2, EnableReplicationUserData uData)
      throws ReplicationCliException
  {
    // Every base DN: a new replication server takes an id that no domain of any backend base DN has (issue #1207).
    // The filter only decides which of the backends and domains read anyway the descriptors keep.
    final TopologyCacheFilter filter = new TopologyCacheFilter();
    filter.setSearchMonitoringInformation(false);
    filter.addBaseDNToSearch(ADSContext.getAdministrationSuffixDN());
    filter.addBaseDNToSearch(Constants.SCHEMA_DN);
    filter.addBaseDNsToSearch(uData.getBaseDNs());

    final ServerDescriptor serverDesc1 = createStandalone(conn1, filter);
    final ServerDescriptor serverDesc2 = createStandalone(conn2, filter);
@@ -4608,7 +4607,7 @@
    if (!argParser.isInteractive())
    {
      // Inform the user of the potential errors that we found in the already registered servers.
      printErrorWhenServersAlreadyHaveErrors(conn1, conn2, uData, adsCtx1, adsCtx2);
      printErrorWhenServersAlreadyHaveErrors(conn1, conn2, adsCtx1, adsCtx2);
    }
    warnIfOnlyOneReplicationServerInTopology(uData, adsCtx1, adsCtx2);

@@ -4767,7 +4766,6 @@
      uData.addBaseDN(Constants.SCHEMA_DN);
    }

    final Set<Integer> usedReplicationServerIds = new HashSet<>();
    final TopologyCache cache1;
    final TopologyCache cache2;
    try
@@ -4775,16 +4773,8 @@
      Set<PreferredConnection> cnx = new LinkedHashSet<>();
      cnx.addAll(getPreferredConnections(conn1));
      cnx.addAll(getPreferredConnections(conn2));
      cache1 = createTopologyCacheOrNull(adsCtx1, cnx, uData);
      if (cache1 != null)
      {
        usedReplicationServerIds.addAll(getReplicationServerIds(cache1));
      }
      cache2 = createTopologyCacheOrNull(adsCtx2, cnx, uData);
      if (cache1 != null)
      {
        usedReplicationServerIds.addAll(getReplicationServerIds(cache1));
      }
      cache1 = createTopologyCacheOrNull(adsCtx1, cnx);
      cache2 = createTopologyCacheOrNull(adsCtx2, cnx);
    }
    catch (ADSContextException adce)
    {
@@ -4799,12 +4789,14 @@
          ERROR_READING_TOPOLOGY_CACHE, tce);
    }

    // The new replication servers and the new domains of every base DN take their ids outside of one set
    final Set<Integer> usedServerIds = getUsedServerIds(cache1, cache2, serverDesc1, serverDesc2);

    final Set<HostPort> twoReplServers = new LinkedHashSet<>();
    addToSets(serverDesc1, uData.getServer1(), conn1, twoReplServers, usedReplicationServerIds);
    addToSets(serverDesc2, uData.getServer2(), conn2, twoReplServers, usedReplicationServerIds);
    addReplicationServer(serverDesc1, uData.getServer1(), conn1, twoReplServers);
    addReplicationServer(serverDesc2, uData.getServer2(), conn2, twoReplServers);

    final Map<DN, Set<HostPort>> hmRepServers = new HashMap<>();
    final Map<DN, Set<Integer>> hmUsedReplicaServerIds = new HashMap<>();
    for (DN baseDN : uData.getBaseDNs())
    {
      Set<HostPort> repServersForBaseDN = new LinkedHashSet<>();
@@ -4812,25 +4804,6 @@
      repServersForBaseDN.addAll(getReplicationServers(baseDN, cache2, serverDesc2));
      repServersForBaseDN.addAll(twoReplServers);
      hmRepServers.put(baseDN, repServersForBaseDN);

      Set<Integer> replicaServerIds = new HashSet<>();
      addReplicaServerIds(replicaServerIds, serverDesc1, baseDN);
      addReplicaServerIds(replicaServerIds, serverDesc2, baseDN);
      if (cache1 != null)
      {
        for (ServerDescriptor server : cache1.getServers())
        {
          addReplicaServerIds(replicaServerIds, server, baseDN);
        }
      }
      if (cache2 != null)
      {
        for (ServerDescriptor server : cache2.getServers())
        {
          addReplicaServerIds(replicaServerIds, server, baseDN);
        }
      }
      hmUsedReplicaServerIds.put(baseDN, replicaServerIds);
    }

    final Set<HostPort> allRepServers = new LinkedHashSet<>();
@@ -4841,22 +4814,21 @@

    Set<String> alreadyConfiguredReplicationServers = new HashSet<>();
    configureServer(conn1, serverDesc1, uData.getServer1(), argParser.server1.replicationPortArg,
        usedReplicationServerIds, allRepServers, alreadyConfiguredReplicationServers,
        usedServerIds, allRepServers, alreadyConfiguredReplicationServers,
        WARN_FIRST_REPLICATION_SERVER_ALREADY_CONFIGURED);
    configureServer(conn2, serverDesc2, uData.getServer2(), argParser.server2.replicationPortArg,
        usedReplicationServerIds, allRepServers, alreadyConfiguredReplicationServers,
        usedServerIds, allRepServers, alreadyConfiguredReplicationServers,
        WARN_SECOND_REPLICATION_SERVER_ALREADY_CONFIGURED);

    for (DN baseDN : uData.getBaseDNs())
    {
      Set<HostPort> repServers = hmRepServers.get(baseDN);
      Set<Integer> usedIds = hmUsedReplicaServerIds.get(baseDN);
      Set<String> alreadyConfiguredServers = new HashSet<>();

      configureToReplicateBaseDN(uData.getServer1(), conn1, serverDesc1, cache1, baseDN,
          usedIds, alreadyConfiguredServers, repServers, allRepServers, alreadyConfiguredReplicationServers);
          usedServerIds, alreadyConfiguredServers, repServers, allRepServers, alreadyConfiguredReplicationServers);
      configureToReplicateBaseDN(uData.getServer2(), conn2, serverDesc2, cache2, baseDN,
          usedIds, alreadyConfiguredServers, repServers, allRepServers, alreadyConfiguredReplicationServers);
          usedServerIds, alreadyConfiguredServers, repServers, allRepServers, alreadyConfiguredReplicationServers);
    }

    // Now that replication is configured in all servers, simply try to
@@ -4930,7 +4902,7 @@
  }

  private void printErrorWhenServersAlreadyHaveErrors(ConnectionWrapper conn1, ConnectionWrapper conn2,
      EnableReplicationUserData uData, final ADSContext adsCtx1, final ADSContext adsCtx2)
      final ADSContext adsCtx1, final ADSContext adsCtx2)
      throws ReplicationCliException
  {
    final Set<LocalizableMessage> messages = new LinkedHashSet<>();
@@ -4939,12 +4911,12 @@
      final Set<PreferredConnection> cnx = new LinkedHashSet<>();
      cnx.addAll(getPreferredConnections(conn1));
      cnx.addAll(getPreferredConnections(conn2));
      final TopologyCache cache1 = createTopologyCacheOrNull(adsCtx1, cnx, uData);
      final TopologyCache cache1 = createTopologyCacheOrNull(adsCtx1, cnx);
      if (cache1 != null)
      {
        messages.addAll(cache1.getErrorMessages());
      }
      final TopologyCache cache2 = createTopologyCacheOrNull(adsCtx2, cnx, uData);
      final TopologyCache cache2 = createTopologyCacheOrNull(adsCtx2, cnx);
      if (cache2 != null)
      {
        messages.addAll(cache2.getErrorMessages());
@@ -5028,22 +5000,33 @@
    }
  }

  private void addReplicaServerIds(Set<Integer> replicaServerIds, ServerDescriptor serverDesc1, DN baseDN)
  /**
   * Returns the server ids that the replication servers and the replication domains of the two
   * servers and of their topologies use, in one set (issue #1207). Only the domains of a backend base
   * DN are in it: the descriptors give a domain id only to the replica of the backend base DN of the
   * domain, so the id of a domain of another base DN, such as one below a backend base DN, is not.
   */
  static Set<Integer> getUsedServerIds(TopologyCache cache1, TopologyCache cache2,
      ServerDescriptor... servers)
  {
    ReplicaDescriptor replica = findReplicatedReplicaForSuffixDN(serverDesc1.getReplicas(), baseDN);
    if (replica != null)
    List<ServerDescriptor> all = new ArrayList<>(Arrays.asList(servers));
    if (cache1 != null)
    {
      replicaServerIds.add(replica.getServerId());
      all.addAll(cache1.getServers());
    }
    if (cache2 != null)
    {
      all.addAll(cache2.getServers());
    }
    return InstallerHelper.getUsedServerIds(all);
  }

  private void addToSets(ServerDescriptor serverDesc, EnableReplicationServerData serverData, ConnectionWrapper conn,
      final Set<HostPort> twoReplServers, final Set<Integer> usedReplicationServerIds)
  private void addReplicationServer(ServerDescriptor serverDesc, EnableReplicationServerData serverData,
      ConnectionWrapper conn, final Set<HostPort> twoReplServers)
  {
    if (serverDesc.isReplicationServer())
    {
      twoReplServers.add(serverDesc.getReplicationServerHostPort());
      usedReplicationServerIds.add(serverDesc.getReplicationServerId());
    }
    else if (serverData.configureReplicationServer())
    {
@@ -5080,7 +5063,7 @@

  private void configureServer(ConnectionWrapper conn, ServerDescriptor serverDesc,
      EnableReplicationServerData enableServer, IntegerArgument replicationPortArg,
      Set<Integer> usedReplicationServerIds, Set<HostPort> allRepServers,
      Set<Integer> usedServerIds, Set<HostPort> allRepServers,
      Set<String> alreadyConfiguredReplicationServers, Arg2<Number, Number> replicationServerAlreadyConfiguredMsg)
      throws ReplicationCliException
  {
@@ -5089,7 +5072,7 @@
      try
      {
        configureAsReplicationServer(conn, enableServer.getReplicationPort(), enableServer.isSecureReplication(),
            allRepServers, usedReplicationServerIds);
            allRepServers, usedServerIds);
      }
      catch (Exception ode)
      {
@@ -5124,15 +5107,15 @@
        ERROR_CONFIGURING_REPLICATIONSERVER, ode);
  }

  private TopologyCache createTopologyCacheOrNull(ADSContext adsCtx, Set<PreferredConnection> cnx,
      ReplicationUserData uData) throws ADSContextException, TopologyCacheException
  private TopologyCache createTopologyCacheOrNull(ADSContext adsCtx, Set<PreferredConnection> cnx)
      throws ADSContextException, TopologyCacheException
  {
    if (adsCtx.hasAdminData())
    {
      TopologyCache cache = new TopologyCache(adsCtx, getTrustManager(sourceServerCI), getConnectTimeout());
      cache.setPreferredConnections(cnx);
      // Every base DN, as for the descriptors of the two servers of enable
      cache.getFilter().setSearchMonitoringInformation(false);
      cache.getFilter().addBaseDNsToSearch(uData.getBaseDNs());
      cache.reloadTopology();
      return cache;
    }
@@ -6164,15 +6147,15 @@
   * the replication port or not.
   * @param replicationServers the list of replication servers to which the
   * replication server will communicate with.
   * @param usedReplicationServerIds the set of replication server IDs that
   * are already in use.  The set will be updated with the replication ID
   * that will be used by the newly configured replication server.
   * @param usedServerIds the set of the server IDs that the replication servers
   * and the replication domains already use.  The set will be updated with the
   * replication ID that will be used by the newly configured replication server.
   * @throws OpenDsException if there is an error updating the configuration.
   */
  private void configureAsReplicationServer(ConnectionWrapper conn,
      int replicationPort, boolean useSecureReplication,
      Set<HostPort> replicationServers,
      Set<Integer> usedReplicationServerIds) throws Exception
      Set<Integer> usedServerIds) throws Exception
  {
    Set<String> replicationServersLC = toLowerCaseStrings(replicationServers);
    print(formatter.getFormattedWithPoints(
@@ -6219,8 +6202,7 @@
        crypto.setSSLEncryption(useSecureReplication);
        crypto.commit();
      }
      int id = InstallerHelper.getReplicationId(usedReplicationServerIds);
      usedReplicationServerIds.add(id);
      int id = InstallerHelper.addNewReplicationId(usedServerIds);
      rsCfgClient = sync.createReplicationServer(
          ReplicationServerCfgDefn.getInstance(),
          new ArrayList<PropertyException>());
@@ -6232,7 +6214,7 @@
    else
    {
      rsCfgClient = sync.getReplicationServer();
      usedReplicationServerIds.add(rsCfgClient.getReplicationServerId());
      usedServerIds.add(rsCfgClient.getReplicationServerId());
      mustCommit = addAllReplicationServers(rsCfgClient, replicationServersLC);
    }
    if (mustCommit)
@@ -6291,26 +6273,6 @@
  }

  /**
   * Returns a Set containing all the replication server ids found in the
   * servers of a given TopologyCache object.
   * @param cache the TopologyCache object to use.
   * @return a Set containing all the replication server ids found in a given
   * TopologyCache object.
   */
  private Set<Integer> getReplicationServerIds(TopologyCache cache)
  {
    Set<Integer> ids = new HashSet<>();
    for (ServerDescriptor server : cache.getServers())
    {
      if (server.isReplicationServer())
      {
        ids.add(server.getReplicationServerId());
      }
    }
    return ids;
  }

  /**
   * Configures a replication domain for a given base DN in the server for which the connection is
   * provided.
   *
@@ -6320,16 +6282,17 @@
   *          the base DN of the replication domain to configure.
   * @param replicationServers
   *          the list of replication servers to which the replication domain will communicate with.
   * @param usedReplicaServerIds
   *          the set of replication domain IDs that are already in use. The set will be updated
   *          with the replication ID that will be used by the newly configured replication server.
   * @param usedServerIds
   *          the set of the server IDs that the replication servers and the replication domains
   *          already use. The set will be updated with the server ID that will be used by the newly
   *          configured replication domain.
   * @throws OpenDsException
   *           if there is an error updating the configuration.
   */
  private void configureToReplicateBaseDN(ConnectionWrapper conn,
      DN baseDN,
      Set<HostPort> replicationServers,
      Set<Integer> usedReplicaServerIds) throws Exception
      Set<Integer> usedServerIds) throws Exception
  {
    Set<String> replicationServersLC = toLowerCaseStrings(replicationServers);

@@ -6362,8 +6325,7 @@
    boolean mustCommit = false;
    if (domain == null)
    {
      int replicaServerId = InstallerHelper.getReplicationId(usedReplicaServerIds);
      usedReplicaServerIds.add(replicaServerId);
      int replicaServerId = InstallerHelper.addNewReplicationId(usedServerIds);
      String domainName = InstallerHelper.getDomainName(domainNames, baseDN);
      domain = sync.createReplicationDomain(
          ReplicationDomainCfgDefn.getInstance(), domainName,
@@ -6436,8 +6398,9 @@
   * provided ServerDescriptor object.
   * @param baseDN the base DN to replicate.
   * @param repServers the replication servers to be defined in the domain.
   * @param usedIds the replication domain Ids already used.  This Set is
   * updated with the new domains that are used.
   * @param usedIds the server Ids that the replication servers and the
   * replication domains already use.  This Set is updated with the new domains
   * that are used.
   * @param cache the TopologyCache used to retrieve the different defined
   * replicas.
   * @param server the ServerDescriptor that is used to identify the
@@ -9047,18 +9010,6 @@
    return null;
  }

  private ReplicaDescriptor findReplicatedReplicaForSuffixDN(Set<ReplicaDescriptor> replicas, DN baseDN)
  {
    for (ReplicaDescriptor replica : replicas)
    {
      if (replica.isReplicated() && replica.getSuffix().getDN().equals(baseDN))
      {
        return replica;
      }
    }
    return null;
  }

  private Set<DN> findAllReplicatedSuffixDNs(Collection<ReplicaDescriptor> replicas)
  {
    Set<DN> results = new HashSet<>();

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -719,6 +719,11 @@
 in domain "%s" is held: a total update is being processed over that session, and stopping it \
 would end the total update. The restart runs once the total update is over, and the change is \
 sent again then; until then it is not recorded as replayed
WARN_REPLICA_HAS_THE_SERVER_ID_OF_A_REPLICATION_SERVER_332=In Replication server %s: directory \
 server DS(%d) of domain "%s" has the server ID of replication server RS(%d). A replication server \
 of an earlier version takes the total update messages for this directory server as messages for \
 itself, and the assured replication counts the acknowledgments of both servers as one: give the \
 domain a server ID that no replication server of the topology uses
ERR_MISSING_CHANGES_NOT_PUBLISHED_333=Domain "%s" could not publish the changes its replication \
 server misses: %s. It tries again within %d ms - on the session which starts meanwhile, or on one it \
 starts then - and the changes done meanwhile are left to that recovery as well

 opendj-server-legacy/src/test/java/org/opends/quicksetup/installer/InstallerHelperTest.java

@@ -20,11 +20,17 @@

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.forgerock.opendj.ldap.Entry;
import org.forgerock.opendj.ldap.LinkedAttribute;
import org.forgerock.opendj.ldap.LinkedHashMapEntry;
import org.opends.admin.ads.ADSContext;
import org.opends.admin.ads.ReplicaDescriptor;
import org.opends.admin.ads.ServerDescriptor;
import org.opends.server.DirectoryServerTestCase;
import org.testng.annotations.Test;

@@ -179,4 +185,68 @@
    assertThat(InstallerHelper.getTaskLogMessages(taskEntry)).isEmpty();
    assertThat(InstallerHelper.getRelevantLogMessage(InstallerHelper.getTaskLogMessages(taskEntry))).isNull();
  }

  /**
   * The server ids in use are the ones of the replication servers and of the replication domains of
   * every base DN, in one set: a new replication server and a new replication domain both take their
   * id outside of it. A replication server takes the total update messages sent to its own id as
   * messages for itself, so a domain must not get the id of a replication server, nor a replication
   * server the id of a domain (issue #1207).
   */
  @Test
  public void usedServerIdsAreTheIdsOfTheReplicationServersAndOfTheDomainsOfEveryBaseDN()
  {
    final ServerDescriptor combined = server(4242, replica(1), replica(2), notReplicatedReplica());
    final ServerDescriptor directoryOnly = server(null, replica(3));
    final ServerDescriptor replicationOnly = server(4243);

    assertThat(InstallerHelper.getUsedServerIds(Arrays.asList(combined, directoryOnly, replicationOnly)))
        .containsOnly(4242, 4243, 1, 2, 3);
  }

  /**
   * A new id is taken outside of the used ids and added to them: the replication server and the
   * domains that setup or enable create next cannot take it (issue #1207). Only 4242 is free.
   */
  @Test
  public void newReplicationIdIsTakenOutsideOfTheUsedIdsAndAddedToThem()
  {
    final Set<Integer> usedIds = new HashSet<>();
    for (int id = 1; id < Short.MAX_VALUE; id++)
    {
      if (id != 4242)
      {
        usedIds.add(id);
      }
    }

    assertThat(InstallerHelper.addNewReplicationId(usedIds)).isEqualTo(4242);
    assertThat(usedIds).contains(4242).hasSize(Short.MAX_VALUE - 1);
  }

  private static ServerDescriptor server(Integer replicationServerId, ReplicaDescriptor... replicas)
  {
    final ServerDescriptor server =
        ServerDescriptor.createStandalone(new HashMap<ADSContext.ServerProperty, Object>());
    if (replicationServerId != null)
    {
      server.getServerProperties().put(ServerDescriptor.ServerProperty.IS_REPLICATION_SERVER, Boolean.TRUE);
      server.getServerProperties().put(ServerDescriptor.ServerProperty.REPLICATION_SERVER_ID, replicationServerId);
    }
    server.setReplicas(new HashSet<>(Arrays.asList(replicas)));
    return server;
  }

  private static ReplicaDescriptor replica(int serverId)
  {
    final ReplicaDescriptor replica = new ReplicaDescriptor();
    replica.setServerId(serverId);
    return replica;
  }

  /** A suffix of the server which is not replicated: its replica has no server id. */
  private static ReplicaDescriptor notReplicatedReplica()
  {
    return new ReplicaDescriptor();
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDomainGenerationIdWarningTest.java

New file
@@ -0,0 +1,55 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

import static org.assertj.core.api.Assertions.*;

import org.opends.server.DirectoryServerTestCase;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

/**
 * Tests how a {@link ReplicationServerDomain} tells the generation id warning of a peer replication
 * server by the id of its message (issue #1207).
 */
@SuppressWarnings("javadoc")
public class ReplicationServerDomainGenerationIdWarningTest extends DirectoryServerTestCase
{
  @DataProvider
  public Object[][] msgIDs()
  {
    return new Object[][] {
      // since OpenDJ 3.0: <resource name>-<ordinal>
      { "org.opends.messages.replication-77", true },
      // OpenDJ 2.6 (ErrorMsg, MessageDescriptor.getId() at tag 2.6.0): SYNC 0x00E00000 | SEVERE_WARNING 0x00020000 | 77
      { "14811213", true },
      // another message of the same file, in either form
      { "org.opends.messages.replication-78", false },
      { "14811214", false },
      // the same ordinal with another severity (SEVERE_ERROR 0x00040000) or in another file
      { "14942285", false },
      { "org.opends.messages.backend-77", false },
      { "77", false },
      { null, false },
    };
  }

  @Test(dataProvider = "msgIDs")
  public void generationIdWarningIsToldByTheIdOfItsMessage(String msgID, boolean isGenerationIdWarning)
  {
    assertThat(ReplicationServerDomain.isGenerationIdWarningMsgID(msgID)).isEqualTo(isGenerationIdWarning);
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java

@@ -13,7 +13,7 @@
 *
 * Copyright 2008-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2025-2026 3A Systems,LLC.
 * Portions Copyright 2025-2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

@@ -24,6 +24,8 @@
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

import java.net.SocketTimeoutException;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.concurrent.BlockingQueue;
@@ -41,11 +43,14 @@
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.protocol.ByteArrayBuilder;
import org.opends.server.replication.protocol.ErrorMsg;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.ServerHandler;
import org.opends.server.replication.service.ReplicationDomain.ImportExportContext;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.tasks.InitializeTask;
@@ -539,6 +544,502 @@
    }
  }

  @DataProvider
  public Object[][] replicaWithTheIdOfItsReplicationServer()
  {
    // the id of the replication server is 11
    return new Object[][] {
      { 1, 11 }, // the importer has it: InitializeTargetMsg and the entries are sent to 11
      { 11, 2 }, // the exporter has it: InitializeRequestMsg and the acknowledgements are sent to 11
    };
  }

  /**
   * A replica whose server id is the one of the replication server it is connected to can be
   * initialized, and can initialize others, through it (issue #1207).
   * <p>
   * The replication server takes a message for one of its replicas, not for itself: it has no data to
   * export or import, and before the fix it answered every total update message sent to its own id
   * with an ErrorMsg, so the total update never got through.
   */
  @Test(dataProvider = "replicaWithTheIdOfItsReplicationServer")
  public void exportAndImportWithAReplicaHavingTheIdOfItsReplicationServer(int exporterId, int importerId)
      throws Exception
  {
    final int ENTRYCOUNT = 100;
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain exporter = null;
    FakeReplicationDomain importer = null;

    try
    {
      int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(11, replServerPort, "replicaWithTheIdOfItsRS", 100);
      SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);

      String exportedData = buildExportedData(ENTRYCOUNT);
      exporter = new FakeReplicationDomain(testService, exporterId, servers, 0, exportedData, null, ENTRYCOUNT);
      StringBuffer importedData = new StringBuffer();
      importer = new FakeReplicationDomain(testService, importerId, servers, 0, null, importedData, 0);
      waitTopologyKnowsReplica(importer, exporterId);
      waitTopologyKnowsReplica(exporter, importerId);

      importer.initializeFromRemote(exporterId, NO_INIT_TASK);

      waitEndExport(exportedData, importedData);
      assertExportSucessful(exporter, importer, exportedData, importedData);
    }
    finally
    {
      disable(exporter, importer);
      remove(replServer);
    }
  }

  @DataProvider
  public Object[][] replicaWithTheIdOfAnotherReplicationServer()
  {
    // the replica with the id of the first replication server, 11, is the importer
    return new Object[][] {
      { 0, 1 }, // a replica of the first RS sends to 11: the first RS hands it on to the second
      { 1, 0 }, // a replica of the second RS sends to 11: the first RS gets it from the second RS
    };
  }

  /**
   * A replica whose server id is the one of a replication server can be initialized across two
   * replication servers, whether it is connected to that replication server or to the other one
   * (issue #1207).
   */
  @Test(dataProvider = "replicaWithTheIdOfAnotherReplicationServer")
  public void exportAndImportAcross2ReplServersWithAReplicaHavingTheIdOfAReplicationServer(
      int exporterRs, int importerRs) throws Exception
  {
    final int ENTRYCOUNT = 100;
    final int exporterId = 1;
    final int importerId = 11;
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    FakeReplicationDomain exporter = null;
    FakeReplicationDomain importer = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAnRS1", 100);
      replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAnRS2", 100, "localhost:" + ports[0]);

      String exportedData = buildExportedData(ENTRYCOUNT);
      exporter = new FakeReplicationDomain(testService, exporterId, newTreeSet("localhost:" + ports[exporterRs]),
          0, exportedData, null, ENTRYCOUNT);
      StringBuffer importedData = new StringBuffer();
      importer = new FakeReplicationDomain(testService, importerId, newTreeSet("localhost:" + ports[importerRs]),
          0, null, importedData, 0);
      waitTopologyKnowsReplica(importer, exporterId);
      waitTopologyKnowsReplica(exporter, importerId);

      importer.initializeFromRemote(exporterId, NO_INIT_TASK);

      waitEndExport(exportedData, importedData);
      assertExportSucessful(exporter, importer, exportedData, importedData);
    }
    finally
    {
      disable(exporter, importer);
      remove(replServer1, replServer2);
    }
  }

  /** A replication server reports a replica which connects to it with its own server id (issue #1207). */
  @Test
  public void replicaWithTheIdOfItsReplicationServerIsReported() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain = null;

    try
    {
      int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(11, replServerPort, "replicaWithTheIdOfItsRSIsReported", 100);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();

      domain = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + replServerPort), 0, 1);

      waitSharedServerIdReported(replServer, 11, testService);
    }
    finally
    {
      disable(domain);
      remove(replServer);
    }
  }

  /**
   * A replication server reports a replica which connects to it with the server id of a replication
   * server it is connected to, and only that replica (issue #1207).
   */
  @Test
  public void replicaWithTheIdOfAConnectedReplicationServerIsReported() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    FakeReplicationDomain domain2 = null;
    FakeReplicationDomain domain11 = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAConnectedRS1", 100);
      replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAConnectedRS2", 100,
          "localhost:" + ports[0]);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();
      // the domain of the second replication server connects it to the first one
      domain2 = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + ports[1]), 0, 1);
      waitTopologyKnowsReplicationServers(domain2, 2);

      domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1);

      waitSharedServerIdReported(replServer2, 11, testService);
      assertEquals(sharedServerIdRecords(replServer2, 2, testService), newArrayList(),
          "a replica whose server id no replication server has must not be reported");
    }
    finally
    {
      disable(domain2, domain11);
      remove(replServer1, replServer2);
    }
  }

  /**
   * A replication server reports a replication server which connects to it with the server id of one
   * of its replicas (issue #1207).
   */
  @Test
  public void replicationServerWithTheIdOfAConnectedReplicaIsReported() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    FakeReplicationDomain domain11 = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer2 = createReplicationServer(12, ports[1], "rsWithTheIdOfAConnectedReplica2", 100);
      domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1);
      waitConnected(domain11);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();

      replServer1 = createReplicationServer(11, ports[0], "rsWithTheIdOfAConnectedReplica1", 100,
          "localhost:" + ports[1]);
      // a replication server connects to its peers for the domains it has
      replServer1.getReplicationServerDomain(testService, true);

      waitSharedServerIdReported(replServer2, 11, testService);
    }
    finally
    {
      disable(domain11);
      remove(replServer1, replServer2);
    }
  }

  /**
   * A replication server takes for itself the warning a peer replication server sends it when their
   * generation ids differ, also when one of its replicas has its server id: the replica would take the
   * warning for the end of the total update it is in (issue #1207).
   */
  @Test
  public void generationIdWarningOfAPeerIsNotForTheReplicaWithTheIdOfTheReplicationServer() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    ReplicationBroker broker11 = null;
    ReplicationBroker broker2 = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer1 = createReplicationServer(11, ports[0], "genIdWarningOfAPeer1", 100);
      replServer2 = createReplicationServer(12, ports[1], "genIdWarningOfAPeer2", 100);
      // a replication server takes the generation id of the first replica which connects to it
      broker11 = openReplicationSession(testService, 11, 100, ports[0], 1000, 1111);
      broker2 = openReplicationSession(testService, 2, 100, ports[1], 1000, 2222);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();

      // the second replication server connects to the first one and warns it that their generation ids differ
      replServer2.applyConfigurationChange(new ReplServerFakeConfiguration(
          ports[1], "genIdWarningOfAPeer2", 0, 12, 0, 100, newTreeSet("localhost:" + ports[0])));

      waitGenerationIdWarningOfPeerReceived(testService, 11, 1111, 12, 2222);
      assertNoErrorMsgReceived(broker11);
    }
    finally
    {
      stop(broker11, broker2);
      remove(replServer1, replServer2);
    }
  }

  @DataProvider
  public Object[][] errorMsgIDsOfAPeer()
  {
    return new Object[][] {
      // the generation id warning, as replication servers send it since OpenDJ 3.0 and as OpenDJ 2.6 sends it
      { "org.opends.messages.replication-77", true },
      { "14811213", true },
      // another message, in either form
      { "org.opends.messages.replication-78", false },
      { "14811214", false },
    };
  }

  /**
   * Of the ErrorMsgs a peer replication server sends to the server id of a replication server, only the
   * generation id warning stays with the replication server, in either form of the id of its message;
   * the replica with that server id gets the others (issue #1207). A replication server of this version
   * sends only the form used since OpenDJ 3.0, so the peer here sends ErrorMsgs decoded from their bytes.
   */
  @Test(dataProvider = "errorMsgIDsOfAPeer")
  public void onlyTheGenerationIdWarningOfAPeerStaysWithTheReplicationServer(String msgID,
      boolean staysWithTheReplicationServer) throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    ReplicationBroker broker11 = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer1 = createReplicationServer(11, ports[0], "errorMsgOfAPeer1", 100);
      replServer2 = createReplicationServer(12, ports[1], "errorMsgOfAPeer2", 100, "localhost:" + ports[0]);
      broker11 = openReplicationSession(testService, 11, 100, ports[0], 1000);
      // a replication server connects to its peers for the domains it has
      replServer2.getReplicationServerDomain(testService, true);
      waitReplicationServerConnected(replServer1, testService, 12);
      waitReplicationServerConnected(replServer2, testService, 11);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();

      ServerHandler toReplServer1 = replServer2.getReplicationServerDomain(testService).getConnectedRSs().get(11);
      String details = "error " + msgID;
      toReplServer1.send(errorMsg(12, 11, msgID, details));
      // sent on the same connection after the first one: the replica gets the first one before it, or never
      String lastDetails = "last error";
      toReplServer1.send(errorMsg(12, 11, "org.opends.messages.replication-78", lastDetails));

      List<String> received = receiveErrorMsgsUntil(broker11, lastDetails);
      if (staysWithTheReplicationServer)
      {
        assertFalse(received.contains(details), "DS(11) received the generation id warning " + msgID);
        waitErrorMsgReceived(details);
      }
      else
      {
        assertTrue(received.contains(details), "DS(11) did not receive the ErrorMsg " + msgID + ": " + received);
      }
    }
    finally
    {
      stop(broker11);
      remove(replServer1, replServer2);
    }
  }

  /**
   * Returns an ErrorMsg that carries the provided id of message, decoded from its bytes as a replication
   * server decodes it: the constructors of ErrorMsg build that id from the details.
   */
  private static ErrorMsg errorMsg(int senderId, int destination, String msgID, String details) throws Exception
  {
    byte[] bytes = new ByteArrayBuilder()
        .appendByte(14) // ReplicationMsg.MSG_TYPE_ERROR
        .appendIntUTF8(senderId)
        .appendIntUTF8(destination)
        .appendString(msgID)
        .appendString(details)
        .appendLongUTF8(System.currentTimeMillis())
        .toByteArray();
    return (ErrorMsg) ReplicationMsg.generateMsg(bytes, ProtocolVersion.getCurrentVersion());
  }

  /** Returns the details of the ErrorMsgs the replica receives before the one with the provided details. */
  private List<String> receiveErrorMsgsUntil(ReplicationBroker broker, String lastDetails) throws Exception
  {
    final List<String> received = newArrayList();
    final long deadline = System.currentTimeMillis() + SECONDS.toMillis(30);
    while (System.currentTimeMillis() < deadline)
    {
      final ReplicationMsg msg;
      try
      {
        msg = broker.receive();
      }
      catch (SocketTimeoutException e)
      {
        continue;
      }
      assertNotNull(msg, "DS(" + broker.getServerId() + ") lost its connection");
      if (msg instanceof ErrorMsg)
      {
        final String details = ((ErrorMsg) msg).getDetails().toString();
        if (details.equals(lastDetails))
        {
          return received;
        }
        received.add(details);
      }
    }
    throw new AssertionError("DS(" + broker.getServerId() + ") did not receive the ErrorMsg '" + lastDetails
        + "', only " + received);
  }

  private void waitReplicationServerConnected(ReplicationServer replServer, DN baseDN, int peerId) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertTrue(
        replServer.getReplicationServerDomain(baseDN, true).getConnectedRSs().containsKey(peerId),
        "RS(" + peerId + ") connected to " + replServer.getMonitorInstanceName()));
  }

  /** Waits for the error log record in which a replication server reports the ErrorMsg it kept. */
  private void waitErrorMsgReceived(String details) throws Exception
  {
    final String received = ERR_ERROR_MSG_RECEIVED.get(details).toString();
    newTimer().repeatUntilSuccess(() ->
    {
      final List<String> records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
      assertTrue(records.stream().anyMatch(record -> record.contains(received)),
          "no replication server reported the ErrorMsg '" + details + "': " + records);
    });
  }

  /**
   * A replication server tells its peers that a replica has left also when the replica has the server
   * id of one of these peers (issue #1207).
   */
  @Test
  public void replicaWithTheIdOfAPeerReplicationServerIsForgottenWhenItLeaves() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer1 = null;
    ReplicationServer replServer2 = null;
    FakeReplicationDomain domain1 = null;
    FakeReplicationDomain domain11 = null;

    try
    {
      int[] ports = TestCaseUtils.findFreePorts(2);
      replServer1 = createReplicationServer(11, ports[0], "replicaWithTheIdOfAPeerLeaves1", 100);
      replServer2 = createReplicationServer(12, ports[1], "replicaWithTheIdOfAPeerLeaves2", 100,
          "localhost:" + ports[0]);
      domain1 = new FakeReplicationDomain(testService, 1, newTreeSet("localhost:" + ports[0]), 0, 1);
      domain11 = new FakeReplicationDomain(testService, 11, newTreeSet("localhost:" + ports[1]), 0, 1);
      waitTopologyKnowsReplica(domain1, 11);

      domain11.disableService();

      waitTopologyForgetsReplica(domain1, 11);
    }
    finally
    {
      disable(domain1, domain11);
      remove(replServer1, replServer2);
    }
  }

  private static TestTimer newTimer()
  {
    return new TestTimer.Builder()
        .maxSleep(30, SECONDS)
        .sleepTimes(100, MILLISECONDS)
        .toTimer();
  }

  private void waitConnected(ReplicationDomain domain) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "DS(" + domain.getServerId() + ") connected"));
  }

  private void waitTopologyForgetsReplica(ReplicationDomain domain, int dsId) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertFalse(domain.getReplicaInfos().containsKey(dsId),
        "DS(" + dsId + ") has left, and DS(" + domain.getServerId() + ") still knows it"));
  }

  private void waitTopologyKnowsReplicationServers(ReplicationDomain domain, int count) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertEquals(domain.getRsInfos().size(), count,
        "replication servers known to DS(" + domain.getServerId() + ")"));
  }

  private void waitSharedServerIdReported(ReplicationServer replServer, int serverId, DN baseDN) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertFalse(sharedServerIdRecords(replServer, serverId, baseDN).isEmpty(),
        replServer.getMonitorInstanceName() + " did not report that DS(" + serverId + ") has the id of an RS"));
  }

  /**
   * Waits for the error log record in which a replication server reports the warning a peer sent it
   * because their generation ids differ.
   */
  private void waitGenerationIdWarningOfPeerReceived(DN baseDN, int rsId, long rsGenerationId, int peerId,
      long peerGenerationId) throws Exception
  {
    // the warning names the address the peer sees the replication server at, which is not known here
    final String[] warning = WARN_BAD_GENERATION_ID_FROM_RS.get(
        rsId, "@", rsGenerationId, baseDN, peerId, peerGenerationId).toString().split("@");
    final String received = ERR_ERROR_MSG_RECEIVED.get(warning[0]).toString();
    newTimer().repeatUntilSuccess(() ->
    {
      for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
      {
        if (record.contains(received) && record.contains(warning[1]))
        {
          return;
        }
      }
      fail("RS(" + rsId + ") did not receive the warning of RS(" + peerId + "): "
          + TestCaseUtils.ERROR_TEXT_WRITER.getMessages());
    });
  }

  private void assertNoErrorMsgReceived(ReplicationBroker broker)
  {
    try
    {
      ReplicationMsg msg;
      while ((msg = broker.receive()) != null)
      {
        assertFalse(msg instanceof ErrorMsg, "DS(" + broker.getServerId() + ") received " + msg);
      }
    }
    catch (SocketTimeoutException expected)
    {
      // nothing more to receive
    }
  }

  /** Returns the records of the error log in which the replication server reports the provided replica. */
  private List<String> sharedServerIdRecords(ReplicationServer replServer, int serverId, DN baseDN)
  {
    final String message = WARN_REPLICA_HAS_THE_SERVER_ID_OF_A_REPLICATION_SERVER.get(
        replServer.getMonitorInstanceName(), serverId, baseDN, serverId).toString();
    final List<String> records = newArrayList();
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    {
      if (record.contains(message))
      {
        records.add(record);
      }
    }
    return records;
  }

  /**
   * When an export requested by a remote replica cannot start (there is no
   * local task reporting the failure), the requester keeps waiting for the

 opendj-server-legacy/src/test/java/org/opends/server/tools/dsreplication/ReplicationCliMainTest.java

New file
@@ -0,0 +1,85 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.tools.dsreplication;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;

import org.opends.admin.ads.ADSContext;
import org.opends.admin.ads.ReplicaDescriptor;
import org.opends.admin.ads.ServerDescriptor;
import org.opends.admin.ads.TopologyCache;
import org.opends.server.DirectoryServerTestCase;
import org.testng.annotations.Test;

@SuppressWarnings("javadoc")
public class ReplicationCliMainTest extends DirectoryServerTestCase
{
  /**
   * The server ids that enable takes the ids of new replication servers and domains outside of are
   * the ones of both topologies and of both servers. The replication server of the second topology
   * was left out before (issue #1207): enable read the first topology twice.
   */
  @Test
  public void usedServerIdsAreTheIdsOfBothTopologiesAndOfBothServers()
  {
    final TopologyCache topology1 = topology(server(null, replica(1)));
    final TopologyCache topology2 = topology(server(4242));

    assertThat(ReplicationCliMain.getUsedServerIds(topology1, topology2, server(4243), server(null, replica(2))))
        .containsOnly(1, 4242, 4243, 2);
  }

  /** A server that is not registered in any administration data has no topology. */
  @Test
  public void usedServerIdsWithoutTopologiesAreTheIdsOfBothServers()
  {
    assertThat(ReplicationCliMain.getUsedServerIds(null, null, server(4243), server(null, replica(2))))
        .containsOnly(4243, 2);
  }

  private static TopologyCache topology(ServerDescriptor... servers)
  {
    final TopologyCache topology = mock(TopologyCache.class);
    when(topology.getServers()).thenReturn(new HashSet<>(Arrays.asList(servers)));
    return topology;
  }

  private static ServerDescriptor server(Integer replicationServerId, ReplicaDescriptor... replicas)
  {
    final ServerDescriptor server =
        ServerDescriptor.createStandalone(new HashMap<ADSContext.ServerProperty, Object>());
    if (replicationServerId != null)
    {
      server.getServerProperties().put(ServerDescriptor.ServerProperty.IS_REPLICATION_SERVER, Boolean.TRUE);
      server.getServerProperties().put(ServerDescriptor.ServerProperty.REPLICATION_SERVER_ID, replicationServerId);
    }
    server.setReplicas(new HashSet<>(Arrays.asList(replicas)));
    return server;
  }

  private static ReplicaDescriptor replica(int serverId)
  {
    final ReplicaDescriptor replica = new ReplicaDescriptor();
    replica.setServerId(serverId);
    return replica;
  }
}