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

Valery Kharseko

23 hours ago 21446c9656ae9cde8f67d1ddb66cd2f99e9a23fb

[#1207] Route total updates to a replica that has the server ID of a replication server, report such a replica, and draw the IDs of new replication servers and domains from one set (#1211)

Fixes #1207

## Summary

A replication server took every routable message sent to its own server
ID as a message for itself. A directory server whose domain had the ID
of a replication server therefore never got the total update messages
sent to it through that replication server: the replication server
answered them with an `ErrorMsg`, and the online initialize of the
domain never got through. `dsreplication enable` and the installer could
give a new domain such an ID, because they drew the IDs of domains and
of replication servers from two separate sets.

The replication server now routes those messages to the directory server
that has the ID, reports such a directory server in its error log, and
`dsreplication enable` and the installer draw the IDs of new replication
servers and of new domains from one set.

## Cause

- `ReplicationServerDomain.process(RoutableMsg, ServerHandler)` first
compared the destination with the ID of the local replication server,
and only when they differed looked for a directory server with that ID.
- `ReplicationCliMain` (enable) checked the ID of a new domain only
against the domains of the same base DN, and the ID of a new replication
server only against the replication servers.
`Installer.configureReplication` kept the same two sets. The second `if
(cache1 != null)` of enable added the replication server IDs of the
first topology twice and those of the second topology never.
- Nothing reported the clash: `isAlreadyConnectedToDS` looks only at the
connected directory servers.

## Change

`ReplicationServerDomain`:

- `process()`: a directory server with the destination ID, connected to
this replication server or behind a replication server connected to it,
gets the message. A replication server holds no data to export or
import, so the total update messages routed by ID are for the directory
server. A message is for this replication server only when no directory
server has the ID - and in one more case: the warning a peer replication
server sends when the generation IDs differ
(`WARN_BAD_GENERATION_ID_FROM_RS`, sent from `receiveTopoInfoFromRS`).
The directory server would take that warning for the end of the total
update it is in, so it stays with the replication server. It is told
from the other messages by the ID of the message, in either form that ID
takes on the wire: `<resource name>-<ordinal>` since OpenDJ 3.0, and the
numeric ID of the message descriptor from an OpenDJ 2.6 replication
server (14811213: category SYNC, severity SEVERE_WARNING, ordinal 77).
- `stopServer()`: the handler, not its server ID, tells whether a
directory server or a replication server is leaving. Before, a directory
server with the ID of a connected replication server went down the
replication server branch of `stopServer()`: it left the directory
servers, but the peers were never sent the new topology, so they never
heard that the directory server had left.
- `register(DataServerHandler)` / `register(ReplicationServerHandler)`:
a directory server that connects with the ID of this replication server
or of a connected replication server, and a replication server that
connects with the ID of a connected directory server, are reported with
the new `WARN_REPLICA_HAS_THE_SERVER_ID_OF_A_REPLICATION_SERVER_332`,
which asks to give the domain an ID that no replication server of the
topology uses.

`InstallerHelper.getUsedServerIds(Collection<ServerDescriptor>)` (new):
the IDs of the replication servers and of the replicated domains of
every backend base DN, in one set.

- `dsreplication enable` fills it from both topology caches and both
servers, `Installer` from its topology cache or from the suffixes it
found, and both take the ID of a new replication server and of a new
domain outside of it (`InstallerHelper.configureReplication` and the
`ReplicationCliMain` methods now take one set).
- enable no longer limits the topology cache and the descriptors of the
two servers to the base DNs it is run for, the administration data and
the schema: its searches read all the backends and domains anyway, and
the filter only dropped those of the other base DNs from the
descriptors, and so from the set.
- A new domain no longer takes the ID of a domain of another base DN
either, which was allowed before; the IDs run from 1 to 32766.
- `InstallerHelper.addNewReplicationId(Set<Integer>)` replaces
`getReplicationId`: it takes the new ID outside of the set and adds it
to the set, so the replication server and the domains that setup or
enable create one after the other get different IDs. The four callers,
the replication server and the domain of setup and of enable, no longer
add the ID themselves.

## Why not refuse the directory server

#1207 proposed that the replication server refuse such a directory
server at the handshake. A domain that has the ID of a replication
server today replicates its changes; only its online initialize fails.
Refusing it would cut those domains off the replication entirely. And
the replication server it connects to refusing it would not be enough:
the replication server with the ID kept the messages its own directory
servers sent to that ID wherever the directory server with the ID was
connected, so every replication server of the topology would have to
refuse it. Routing the messages to the directory server and reporting it
repairs the existing clashes instead, and the server ID of a domain,
which is read-only, can be changed when convenient.

## Not changed

- A replication server of an earlier version in the same topology still
keeps the messages for a directory server with its own ID; the warning
tells the administrator to give the domain another ID.
- The assured replication keys the expected acknowledgments by server
ID, so it still counts the acknowledgments of a replication server and
of a directory server with the same ID as one. The warning and the
javadoc of `getUsedServerIds` say so.
- `getUsedServerIds` holds only what the descriptors carry: a server
that `dsreplication` or the installer has no descriptor of (not
registered in the administration data, or not reachable) is not in the
set.
- Nor is a domain whose base DN is not the base DN of a backend, such as
a domain below a backend base DN, which
`LDAPReplicationDomain.isConfigurationAcceptable` accepts:
`ServerDescriptor` gives a domain ID only to the replica of the backend
base DN of the domain. Master had the same gap; whether such a domain
can clash in practice was not traced. The comment in
`updateConfiguration` and the javadoc of both `getUsedServerIds` say so.
- The warning is logged each time such a directory server connects.
- The form of the generation ID warning of OpenDJ 2.6 is read from the
sources of tag `2.6.0` (`ErrorMsg`, `MessageDescriptor.getId()`); no 2.6
peer was run, and OpenDJ 2.0-2.4, whose sources are not in this
repository, were not checked.

## Tests

`ReplicationDomainTest`:

| Case | What it pins | master |
|---|---|---|
| `exportAndImportWithAReplicaHavingTheIdOfItsReplicationServer`
(importer 11, exporter 11) | A replica with the ID of its own
replication server imports and exports through it. | red |
|
`exportAndImportAcross2ReplServersWithAReplicaHavingTheIdOfAReplicationServer`
(sender on either replication server) | The same through two replication
servers, the replica with the ID of the first one connected to the
second, and the other way round. | red |
| `replicaWithTheIdOfItsReplicationServerIsReported` | The warning when
the directory server has the ID of its replication server. | red |
| `replicaWithTheIdOfAConnectedReplicationServerIsReported` | The
warning when it has the ID of a peer of its replication server. | red |
| `replicationServerWithTheIdOfAConnectedReplicaIsReported` | The
warning when a replication server connects with the ID of a connected
directory server. | red |
|
`generationIdWarningOfAPeerIsNotForTheReplicaWithTheIdOfTheReplicationServer`
| The generation ID warning of a peer stays with the replication server,
and the directory server with its ID gets no `ErrorMsg`. | green |
| `onlyTheGenerationIdWarningOfAPeerStaysWithTheReplicationServer`
(message ID of the warning since OpenDJ 3.0, of OpenDJ 2.6, of another
message in either form) | A peer sends an `ErrorMsg` to the ID of the
replication server, a directory server with that ID connected: the
warning in either form stays with the replication server, which logs it,
and the directory server gets any other one. The peer sends `ErrorMsg`s
decoded from bytes, as only those carry a message ID of its choice, and
a last one after each: the directory server gets the first one before
the last one, or never. | red (every row: the replication server keeps
every `ErrorMsg` sent to its ID, the last one too) |
| `replicaWithTheIdOfAPeerReplicationServerIsForgottenWhenItLeaves` |
When a directory server with the ID of a peer replication server leaves,
the other replicas stop seeing it. | red |


`InstallerHelperTest.usedServerIdsAreTheIdsOfTheReplicationServersAndOfTheDomainsOfEveryBaseDN`
pins the set: replication servers and replicated domains of every base
DN, not the domains which are not replicated (new method).


`InstallerHelperTest.newReplicationIdIsTakenOutsideOfTheUsedIdsAndAddedToThem`
pins `addNewReplicationId`: with one ID left free, it returns that ID
and adds it to the set (new method).

`ReplicationServerDomainGenerationIdWarningTest` (new) pins the ID that
tells the generation ID warning of a peer: both forms of it, and not the
ID of another message in either form, nor ordinal 77 with another
severity or of another message file.

`ReplicationCliMainTest` (new) pins
`ReplicationCliMain.getUsedServerIds` (now package-private), the helper
enable fills the set with: the replication servers and domains of both
topology caches and of both servers, and only those of the two servers
when there is no topology cache. The topology caches are Mockito stubs.
It calls the helper directly: the arguments that enable passes to it are
not pinned (see below).

The generation ID case, green on master, pins what the change keeps; it
fails if `process()` only looks for a directory server first. Mutants,
each run against the cases above:

| Mutant | Caught by |
|---|---|
| no warning for the ID of the local replication server |
`replicaWithTheIdOfItsReplicationServerIsReported` |
| no warning for the ID of a connected replication server |
`replicaWithTheIdOfAConnectedReplicationServerIsReported` |
| no warning in `register(ReplicationServerHandler)` |
`replicationServerWithTheIdOfAConnectedReplicaIsReported` |
| the warning for every directory server |
`replicaWithTheIdOfAConnectedReplicationServerIsReported` (a replica
with an ID no replication server has is not reported) |
| `process()` checks the local ID first, as on master | the four export
and import cases |
| `process()` looks for a directory server first, no exception for the
warning | `generationIdWarningOfAPeer…`, `onlyTheGenerationIdWarning…`
(both forms of the warning) |
| `stopServer()` classifies by server ID, as on master |
`replicaWithTheIdOfAPeerReplicationServerIsForgottenWhenItLeaves` |
| `getUsedServerIds` without the replication servers / without the
domains / with the domains which are not replicated |
`usedServerIdsAre…` |
| `isGenerationIdWarningOfAPeer` without the check of the message ID |
`onlyTheGenerationIdWarning…` (every row: no other `ErrorMsg` of the
peer reaches the directory server) |
| `isGenerationIdWarningOfAPeer` checks only the form since 3.0, as
before review round 1 | `onlyTheGenerationIdWarning…` (OpenDJ 2.6 form)
|
| `isGenerationIdWarningMsgID` without the OpenDJ 2.6 form / without the
form since 3.0 | `generationIdWarningIsToldByTheIdOfItsMessage`; without
the 2.6 form, also `onlyTheGenerationIdWarning…` |
| `getUsedServerIds` reads the first topology cache twice / the second
one twice / leaves out the two servers |
`usedServerIdsAreTheIdsOfBothTopologiesAndOfBothServers` |
| `updateConfiguration` passes the first topology cache twice, as on
master, or leaves out a server | **not caught**: no test runs `enable` |
| `addNewReplicationId` does not add the ID / ignores the used IDs |
`newReplicationIdIsTakenOutsideOfTheUsedIdsAndAddedToThem` |

No test runs `enable` or `InstallerHelper.configureReplication` against
servers. Not pinned, as all three need `enable` against two servers: the
arguments `updateConfiguration` passes to `getUsedServerIds`
(`ReplicationCliMain.java:4793`), enable limiting the descriptors to its
base DNs again (the filters of `updateConfiguration` and
`createTopologyCacheOrNull`), and enable giving each base DN a set of
its own.

## Verification

- `ReplicationDomainTest` and `InstallerHelperTest`: 35/35.
- `-Pprecommit verify -pl opendj-server-legacy -am` (javadoc included)
with `org/opends/server/replication/**` and `org/opends/quicksetup/**`:
3971 tests, 0 failures.
- Review round 1, on master with #1205 and #1212: `-Pprecommit verify
-pl opendj-server-legacy -am` (javadoc included) with
`ReplicationDomainTest`, `InstallerHelperTest`,
`ReplicationServerDomainGenerationIdWarningTest` and
`ReplicationCliMainTest`: 46/46. The 13 classes of
`org/opends/quicksetup/**`, `SuffixDescriptorTest`,
`ArgumentParserToolsTestCase` and the two new classes, each in a JVM of
its own: 170 tests, one failure, `AdsTrustStoreInstallTestCase`, whose
server could not bind the administration port 65534 that another server
on the machine held; 6/6 when run again. The 7 new mutants are each
caught by their case.
- Review round 2, on master with #1217, #1214 and #1192: `-Pprecommit
verify -pl opendj-server-legacy -am` with the same four classes built
the module and its javadoc, and passed `InstallerHelperTest` (13) and
`ReplicationServerDomainGenerationIdWarningTest` (8);
`ReplicationDomainTest` could not start its server, as another server on
the machine held the administration port 65534, and the run was stopped
there. On the classes of that build, each in a JVM of its own:
`ReplicationDomainTest` 27/27 (at the second attempt, the first one
having lost port 65534 again) and `ReplicationCliMainTest` 2/2. The new
case on master's `ReplicationServerDomain`: 4 of 4 rows red. Four
mutants each fail it: `isGenerationIdWarningOfAPeer` checking only the
form since 3.0 (the OpenDJ 2.6 row), `isGenerationIdWarningOfAPeer`
without the check of the message ID (every row), `process()` without the
exception for the warning (both rows of the warning),
`isGenerationIdWarningMsgID` without the OpenDJ 2.6 form (the OpenDJ 2.6
row).
- The reproduction of #1207 on a package of this branch, before the
changes to `stopServer()`, to the generation ID warning and to the
filter of enable: the domain of B with server ID 4242, the replication
server of A with replication server ID 4242, `dsreplication initialize`
ends in 4 seconds and B holds the 22 entries of A (on master: never
ends).
- The Docker replication test (`docker-test-replication.sh`) was not run
locally.

 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,17 +1622,33 @@
    }
    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);
      }
    }

  /**
   * 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);
  }

  /**
@@ -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

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