From 21446c9656ae9cde8f67d1ddb66cd2f99e9a23fb Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 19 Oct 2026 08:38:00 +0000
Subject: [PATCH] [#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)
---
opendj-server-legacy/src/main/java/org/opends/server/tools/dsreplication/ReplicationCliMain.java | 159 ++-----
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 503 ++++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/quicksetup/installer/InstallerHelperTest.java | 70 +++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDomainGenerationIdWarningTest.java | 55 ++
opendj-server-legacy/src/main/java/org/opends/admin/ads/ReplicaDescriptor.java | 3
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 5
opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/Installer.java | 34 -
opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/InstallerHelper.java | 69 ++
opendj-server-legacy/src/test/java/org/opends/server/tools/dsreplication/ReplicationCliMainTest.java | 85 ++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 109 ++++-
10 files changed, 923 insertions(+), 169 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/admin/ads/ReplicaDescriptor.java b/opendj-server-legacy/src/main/java/org/opends/admin/ads/ReplicaDescriptor.java
index 61a1854..7b66529 100644
--- a/opendj-server-legacy/src/main/java/org/opends/admin/ads/ReplicaDescriptor.java
+++ b/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;
diff --git a/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/Installer.java b/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/Installer.java
index 09ab444..a67b651 100644
--- a/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/Installer.java
+++ b/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
diff --git a/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/InstallerHelper.java b/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/InstallerHelper.java
index 48cdf03..1a99378 100644
--- a/opendj-server-legacy/src/main/java/org/opends/quicksetup/installer/InstallerHelper.java
+++ b/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;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
index c025189..1ed61e5 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
+++ b/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)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/tools/dsreplication/ReplicationCliMain.java b/opendj-server-legacy/src/main/java/org/opends/server/tools/dsreplication/ReplicationCliMain.java
index 4dbb239..78bf317 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/tools/dsreplication/ReplicationCliMain.java
+++ b/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<>();
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index 5b62db3..eb763ec 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/quicksetup/installer/InstallerHelperTest.java b/opendj-server-legacy/src/test/java/org/opends/quicksetup/installer/InstallerHelperTest.java
index c4d28ab..1441455 100644
--- a/opendj-server-legacy/src/test/java/org/opends/quicksetup/installer/InstallerHelperTest.java
+++ b/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();
+ }
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDomainGenerationIdWarningTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDomainGenerationIdWarningTest.java
new file mode 100644
index 0000000..b9c9254
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDomainGenerationIdWarningTest.java
@@ -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);
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
index 37569cb..62b9927 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/tools/dsreplication/ReplicationCliMainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/tools/dsreplication/ReplicationCliMainTest.java
new file mode 100644
index 0000000..ca1fb92
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/tools/dsreplication/ReplicationCliMainTest.java
@@ -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;
+ }
+}
--
Gitblit v1.10.0