From 5055b6312ff61ef861be42bfb2827aa561bc1593 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 29 Oct 2026 09:20:25 +0000
Subject: [PATCH] [#1235] [#1236] [#1240] Replication: no mismatch warning for a peer without a generation ID, and tell the peers the one taken from a TopologyMsg or an update (#1237)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 456 ++++++++++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 56 +++++
2 files changed, 506 insertions(+), 6 deletions(-)
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 1ed61e5..69d3cc4 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
@@ -327,7 +327,16 @@
{
sourceHandler.updateServerState(updateMsg);
sourceHandler.incrementInCount();
- setGenerationIdIfUnset(sourceHandler.getGenerationId());
+ /*
+ * A directory server which already has changes does not give its generation id at the handshake to a
+ * replication server which has none (DataServerHandler.waitAndProcessStartSessionFromRemoteDS()): its first
+ * update does. The peers learn it only from a TopologyMsg, and forward no update to this replication server
+ * while they hold -1 for it, so it is told to them, as in receiveTopoInfoFromRS() (issue #1240).
+ */
+ if (setGenerationIdIfUnset(sourceHandler.getGenerationId()))
+ {
+ sendTopoInfoToAll();
+ }
/**
* If this is an assured message (a message requesting ack), we must
@@ -2398,13 +2407,45 @@
if (allowResetGenId)
{
resetGenerationIdIfPossible();
- setGenerationIdIfUnset(rsHandler.getGenerationId());
+ /*
+ * The peers learn the generation id taken here only from a TopologyMsg, and forward no update to a
+ * replication server while they hold another one for it, so it is told to them as changeGenerationId()
+ * tells its own (issue #1236).
+ *
+ * It is taken only from a peer which has directory servers. A peer without any can reset the one it
+ * presented on the next TopologyMsg it receives, and two replication servers taking such a generation
+ * id from each other and telling each other would go on resetting it and taking it again in turn. Such
+ * a peer has no update of a directory server to forward here anyway, and the handshake still takes its
+ * generation id, see ReplicationServerHandler.checkGenerationId().
+ *
+ * Nor is it taken while this replication server has directory servers of its own. With them connected,
+ * resetGenerationIdIfPossible() never resets the generation id here: this one has none because a reset
+ * of the generation id asked for it, such as the one of dsreplication pre-external-initialization, or
+ * because its directory servers have changes and gave none at the handshake, and put() takes the one
+ * of their first update. A TopologyMsg which the peer sent before it got such a reset still presents
+ * the generation id the reset cleared: taking it would undo the reset here and, told to the peers,
+ * on them as well, for good while the directory servers stay connected. A replication server without
+ * directory servers which takes it resets it again on the TopologyMsg the peer sends once reset.
+ */
+ if (connectedDSs.isEmpty() && rsHandler.hasRemoteLDAPServers()
+ && setGenerationIdIfUnset(rsHandler.getGenerationId()))
+ {
+ sendTopoInfoToAll();
+ }
}
- if (isDifferentGenerationId(rsHandler.getGenerationId()))
+ /*
+ * A peer which presents no generation id has none to differ from this one - the handshake already
+ * holds this, see ReplicationServerHandler.checkGenerationId(). It reset its own, either seeing no
+ * directory server left in the topology or told to by a reset of the generation id, such as the one
+ * of dsreplication pre-external-initialization. Warning about it would ask for a re-initialization
+ * every time the directory servers of a domain which had no change all stop (issue #1235).
+ */
+ final long peerGenerationId = rsHandler.getGenerationId();
+ if (peerGenerationId > 0 && isDifferentGenerationId(peerGenerationId))
{
LocalizableMessage message = WARN_BAD_GENERATION_ID_FROM_RS.get(rsHandler.getServerId(),
- rsHandler.session.getReadableRemoteAddress(), rsHandler.getGenerationId(),
+ rsHandler.session.getReadableRemoteAddress(), peerGenerationId,
baseDN, getLocalRSServerId(), generationId);
logger.warn(message);
@@ -2430,14 +2471,17 @@
}
}
- private void setGenerationIdIfUnset(long generationId)
+ /** Takes the provided generation id when this domain has none, and returns whether it took it. */
+ private boolean setGenerationIdIfUnset(long generationId)
{
synchronized (generationIDLock)
{
- if (this.generationId < 0)
+ if (this.generationId < 0 && this.generationId != generationId)
{
this.generationId = generationId;
+ return true;
}
+ return false;
}
}
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 62b9927..16b07f6 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
@@ -25,6 +25,7 @@
import static org.testng.Assert.*;
import java.net.SocketTimeoutException;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
@@ -39,17 +40,26 @@
import org.opends.server.TestCaseUtils;
import org.opends.server.backends.task.Task;
import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.DSInfo;
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.plugin.DomainFakeCfg;
+import org.opends.server.replication.plugin.DummyReplicationDomain;
import org.opends.server.replication.protocol.ByteArrayBuilder;
+import org.opends.server.replication.protocol.DeleteMsg;
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.ResetGenerationIdMsg;
+import org.opends.server.replication.protocol.TopologyMsg;
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.ReplicationServerDomain;
+import org.opends.server.replication.server.ReplicationServerHandler;
import org.opends.server.replication.server.ServerHandler;
import org.opends.server.replication.service.ReplicationDomain.ImportExportContext;
import org.forgerock.opendj.ldap.DN;
@@ -786,6 +796,452 @@
}
}
+ /**
+ * A peer replication server which presents no generation id has reset its own, seeing no replica left in
+ * the topology: it has none to differ from, and a replication server which still has one does not warn
+ * about it, nor send it the warning (issue #1235). A peer which presents another generation id is still
+ * warned about.
+ */
+ @Test
+ public void peerWithoutGenerationIdIsNotWarnedAbout() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(21, ports[0], "peerWithoutGenerationId1", 100);
+ replServer2 = createReplicationServer(22, ports[1], "peerWithoutGenerationId2", 100, "localhost:" + ports[0]);
+ // the first replication server takes the generation id of the replica, which keeps it from resetting it,
+ // and the second one takes it from the first
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 3333);
+ replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 22);
+ waitReplicationServerConnected(replServer2, testService, 21);
+
+ // what the second replication server sends once it has reset its generation id
+ ServerHandler toReplServer1 = replServer2.getReplicationServerDomain(testService).getConnectedRSs().get(21);
+ toReplServer1.send(topologyMsgOf(replServer2, -1));
+ // sent on the same connection after it: once the warning about this one is back, the one before was handled
+ toReplServer1.send(topologyMsgOf(replServer2, 4444));
+
+ waitGenerationIdWarningOfPeerReceived(testService, 22, 4444L, 21, 3333L);
+ // the warning, as the first replication server logs it or as the second one logs it on receipt
+ final String[] warning = WARN_BAD_GENERATION_ID_FROM_RS.get(
+ 22, "@", -1L, testService, 21, 3333L).toString().split("@");
+ for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
+ {
+ assertFalse(record.contains(warning[0]) && record.contains(warning[1]),
+ "RS(22) without a generation id was taken for a mismatch: " + record);
+ }
+ }
+ finally
+ {
+ stop(broker1);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A replication server which takes the generation id of a peer from its TopologyMsg, after it reset its own,
+ * tells its peers, and they forward it the updates again (issue #1236).
+ */
+ @Test
+ public void peerWhichTakesTheGenerationIdAgainIsForwardedUpdates() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+ ReplicationBroker broker2 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(41, ports[0], "takesTheGenerationIdAgain1", 100);
+ replServer2 = createReplicationServer(42, ports[1], "takesTheGenerationIdAgain2", 100, "localhost:" + ports[0]);
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 5555);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 42);
+ waitReplicationServerConnected(replServer2, testService, 41);
+ waitGenerationIdOfPeer(replServer1, testService, 42, 5555);
+ // no TopologyMsg of the first one for its replica is left to cross what follows
+ waitReplicaOfPeer(replServer2, testService, 41, 5555);
+
+ // what the second replication server does when it sees no replica left in the topology
+ domain2.changeGenerationId(-1);
+ waitGenerationIdOfPeer(replServer1, testService, 42, -1);
+
+ // a replica connecting to the first replication server makes it send its TopologyMsg to its peers
+ broker2 = openReplicationSession(testService, 2, 100, ports[0], 1000, 5555);
+ waitGenerationId(replServer2, testService, 5555);
+ waitGenerationIdOfPeer(replServer1, testService, 42, 5555);
+
+ CSN csn = new CSNGenerator(1, 0).newCSN();
+ broker1.publish(new DeleteMsg(DN.valueOf("cn=deleted," + testService), csn, "uuid"));
+ newTimer().repeatUntilSuccess(() -> assertTrue(domain2.getLatestServerState().cover(csn),
+ "RS(42) did not get the update " + csn + " of DS(1)"));
+ }
+ finally
+ {
+ stop(broker1, broker2);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * Two replication servers which have no replica and the same generation id, which they did not save, send
+ * each other a TopologyMsg with it: both end without a generation id and stay there (issue #1236).
+ */
+ @Test
+ public void peersWithoutReplicaEndWithoutGenerationId() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(51, ports[0], "peersWithoutReplica1", 100);
+ replServer2 = createReplicationServer(52, ports[1], "peersWithoutReplica2", 100, "localhost:" + ports[0]);
+ ReplicationServerDomain domain1 = replServer1.getReplicationServerDomain(testService, true);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 52);
+ waitReplicationServerConnected(replServer2, testService, 51);
+
+ // the generation id the last replicas left them, each telling the other
+ domain1.changeGenerationId(6666);
+ domain2.changeGenerationId(6666);
+
+ waitGenerationId(replServer1, testService, -1);
+ waitGenerationId(replServer2, testService, -1);
+ // and they stay there
+ final long end = System.currentTimeMillis() + SECONDS.toMillis(2);
+ while (System.currentTimeMillis() < end)
+ {
+ assertEquals(domain1.getGenerationId(), -1L, "generation id of RS(51)");
+ assertEquals(domain2.getGenerationId(), -1L, "generation id of RS(52)");
+ Thread.sleep(10);
+ }
+ }
+ finally
+ {
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A replication server without a generation id does not take the one of a peer which has no replica: that
+ * peer can reset it on the next TopologyMsg it receives, and two replication servers taking it from each
+ * other and telling each other would go on resetting it and taking it again in turn (issue #1236).
+ */
+ @Test
+ public void peerWithoutReplicaGivesNoGenerationId() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(61, ports[0], "peerWithoutReplicaGivesNoGenerationId1", 100);
+ replServer2 = createReplicationServer(62, ports[1], "peerWithoutReplicaGivesNoGenerationId2", 100,
+ "localhost:" + ports[0]);
+ ReplicationServerDomain domain1 = replServer1.getReplicationServerDomain(testService, true);
+ replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 62);
+ waitReplicationServerConnected(replServer2, testService, 61);
+
+ // what the second replication server sends once it took a generation id which it has not saved, handled
+ // here rather than by the reader of the connection: what this replication server would tell the second one
+ // could only come back after the assertion
+ ReplicationServerHandler toReplServer2 = domain1.getConnectedRSs().get(62);
+ domain1.receiveTopoInfoFromRS(topologyMsgOf(replServer2, 6666), toReplServer2, true);
+
+ assertEquals(toReplServer2.getGenerationId(), 6666L, "generation id RS(61) has for RS(62)");
+ assertEquals(domain1.getGenerationId(), -1L, "generation id of RS(61)");
+ }
+ finally
+ {
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A replication server which resets its generation id on the TopologyMsg of a peer which has a replica and
+ * another generation id takes the one of that peer in the same call, and tells its peers (issue #1236).
+ */
+ @Test
+ public void peerWithReplicaGivesItsGenerationIdRightAfterAReset() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(71, ports[0], "givesItsGenerationIdRightAfterAReset1", 100);
+ replServer2 = createReplicationServer(72, ports[1], "givesItsGenerationIdRightAfterAReset2", 100,
+ "localhost:" + ports[0]);
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 5555);
+ ReplicationServerDomain domain1 = replServer1.getReplicationServerDomain(testService);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 72);
+ waitReplicationServerConnected(replServer2, testService, 71);
+ // no TopologyMsg of the first one for its replica is left to cross what follows
+ waitReplicaOfPeer(replServer2, testService, 71, 5555);
+
+ // a generation id which the second replication server has not saved and no replica has
+ domain2.changeGenerationId(7777);
+ waitGenerationIdOfPeer(replServer1, testService, 72, 7777);
+
+ // sent once, so that the second replication server can only take the generation id in the call that resets it
+ domain1.getConnectedRSs().get(72).send(domain1.createTopologyMsgForRS());
+ waitGenerationId(replServer2, testService, 5555);
+ waitGenerationIdOfPeer(replServer1, testService, 72, 5555);
+ }
+ finally
+ {
+ stop(broker1);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A TopologyMsg which a peer sent before it got a reset of the generation id, as dsreplication
+ * pre-external-initialization asks a replica for, still presents the generation id the reset cleared: a
+ * replication server which has replicas does not take it, which would undo the reset there and on its peers
+ * for good, nothing resetting it again while the replicas stay connected (issue #1236).
+ */
+ @Test
+ public void topologyMsgSentBeforeAResetDoesNotUndoIt() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+ ReplicationBroker broker2 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(81, ports[0], "topologyMsgSentBeforeAReset1", 100);
+ replServer2 = createReplicationServer(82, ports[1], "topologyMsgSentBeforeAReset2", 100,
+ "localhost:" + ports[0]);
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 5555);
+ broker2 = openReplicationSession(testService, 2, 100, ports[1], 1000, 5555);
+ ReplicationServerDomain domain1 = replServer1.getReplicationServerDomain(testService);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService);
+ waitReplicationServerConnected(replServer1, testService, 82);
+ waitReplicationServerConnected(replServer2, testService, 81);
+ // once each one knows the replica of the other, no TopologyMsg is left to cross the reset
+ waitReplicaOfPeer(replServer1, testService, 82, 5555);
+ waitReplicaOfPeer(replServer2, testService, 81, 5555);
+ final TopologyMsg sentBeforeTheReset = domain2.createTopologyMsgForRS();
+
+ // what dsreplication pre-external-initialization asks a replica for
+ broker1.publish(new ResetGenerationIdMsg(-1));
+ waitGenerationId(replServer1, testService, -1);
+ waitGenerationId(replServer2, testService, -1);
+ waitReplicaOfPeer(replServer1, testService, 82, -1);
+
+ // handled here rather than by the reader of the connection, as when it crosses the reset
+ domain1.receiveTopoInfoFromRS(sentBeforeTheReset, domain1.getConnectedRSs().get(82), true);
+ assertEquals(domain1.getGenerationId(), -1L, "generation id of RS(81)");
+ }
+ finally
+ {
+ stop(broker1, broker2);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A replication server without replicas and without a generation id, to which a peer with replicas presents
+ * none either, as dsreplication pre-external-initialization leaves them, has nothing to take, and tells its
+ * peers nothing (issue #1236).
+ */
+ @Test
+ public void peerWithReplicaAndWithoutGenerationIdIsNotAnswered() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(101, ports[0], "peerWithReplicaIsNotAnswered1", 100);
+ replServer2 = createReplicationServer(102, ports[1], "peerWithReplicaIsNotAnswered2", 100,
+ "localhost:" + ports[0]);
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 5555);
+ ReplicationServerDomain domain1 = replServer1.getReplicationServerDomain(testService);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 102);
+ waitReplicationServerConnected(replServer2, testService, 101);
+ waitReplicaOfPeer(replServer2, testService, 101, 5555);
+
+ // what dsreplication pre-external-initialization asks a replica for
+ broker1.publish(new ResetGenerationIdMsg(-1));
+ waitGenerationId(replServer1, testService, -1);
+ waitGenerationId(replServer2, testService, -1);
+ waitGenerationIdOfPeer(replServer1, testService, 102, -1);
+ waitReplicaOfPeer(replServer2, testService, 101, -1);
+
+ broker1.setSoTimeout(100);
+ // the TopologyMsgs of the reset itself
+ countTopologyMsgs(broker1, SECONDS.toMillis(1));
+ // what the first replication server sends its peers, handled here rather than by the reader of the
+ // connection: whatever the second one tells the first one reaches the replica as a TopologyMsg
+ domain2.receiveTopoInfoFromRS(domain1.createTopologyMsgForRS(), domain2.getConnectedRSs().get(101), true);
+ final int received = countTopologyMsgs(broker1, SECONDS.toMillis(2));
+ // a closed session receives nothing, which would count as quiet
+ assertTrue(broker1.isConnected(), "DS(1) is still connected");
+ assertEquals(received, 0, "TopologyMsgs DS(1) received once RS(102) was presented no generation id");
+ }
+ finally
+ {
+ stop(broker1);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /**
+ * A replication server without a generation id, which takes the one of a replica that already has changes
+ * from the first update of that replica, tells its peers, and they forward it the updates again
+ * (issue #1240).
+ */
+ @Test
+ public void generationIdTakenFromAnUpdateIsToldToThePeers() throws Exception
+ {
+ DN testService = DN.valueOf("o=test");
+ ReplicationServer replServer1 = null;
+ ReplicationServer replServer2 = null;
+ ReplicationBroker broker1 = null;
+ ReplicationBroker broker2 = null;
+
+ try
+ {
+ int[] ports = TestCaseUtils.findFreePorts(2);
+ replServer1 = createReplicationServer(91, ports[0], "generationIdTakenFromAnUpdate1", 100);
+ replServer2 = createReplicationServer(92, ports[1], "generationIdTakenFromAnUpdate2", 100,
+ "localhost:" + ports[0]);
+ broker1 = openReplicationSession(testService, 1, 100, ports[0], 1000, 5555);
+ ReplicationServerDomain domain2 = replServer2.getReplicationServerDomain(testService, true);
+ waitReplicationServerConnected(replServer1, testService, 92);
+ waitReplicationServerConnected(replServer2, testService, 91);
+ // no TopologyMsg of the first one for its replica is left to cross what follows
+ waitReplicaOfPeer(replServer2, testService, 91, 5555);
+
+ domain2.changeGenerationId(-1);
+ waitGenerationIdOfPeer(replServer1, testService, 92, -1);
+
+ // a replica which has changes does not give its generation id at the handshake
+ CSNGenerator csnGenerator2 = new CSNGenerator(2, 0);
+ ServerState state2 = new ServerState();
+ state2.update(csnGenerator2.newCSN());
+ broker2 = openReplicationSession(testService, 2, ports[1], state2, 5555);
+ // the TopologyMsg the second replication server sent for the replica is handled, and nothing is left
+ // to tell the first one the generation id but the update
+ waitReplicaOfPeer(replServer1, testService, 92, -1);
+ assertEquals(domain2.getGenerationId(), -1L, "generation id of RS(92) before the update of DS(2)");
+
+ broker2.publish(new DeleteMsg(DN.valueOf("cn=deleted2," + testService), csnGenerator2.newCSN(), "uuid2"));
+ waitGenerationId(replServer2, testService, 5555);
+ waitGenerationIdOfPeer(replServer1, testService, 92, 5555);
+
+ CSN csn1 = new CSNGenerator(1, 0).newCSN();
+ broker1.publish(new DeleteMsg(DN.valueOf("cn=deleted1," + testService), csn1, "uuid1"));
+ newTimer().repeatUntilSuccess(() -> assertTrue(domain2.getLatestServerState().cover(csn1),
+ "RS(92) did not get the update " + csn1 + " of DS(1)"));
+ }
+ finally
+ {
+ stop(broker1, broker2);
+ remove(replServer1, replServer2);
+ }
+ }
+
+ /** Opens a session of a replica which announces the provided server state rather than an empty one. */
+ private ReplicationBroker openReplicationSession(DN baseDN, int serverId, int port, ServerState state,
+ long generationId) throws Exception
+ {
+ final DomainFakeCfg config = newFakeCfg(baseDN, serverId, port);
+ config.setWindowSize(100);
+ final ReplicationBroker broker = new ReplicationBroker(
+ new DummyReplicationDomain(generationId), state, config, getReplSessionSecurity());
+ connect(broker, 1000);
+ return broker;
+ }
+
+ /** Returns how many TopologyMsgs the broker receives in the provided time. */
+ private static int countTopologyMsgs(ReplicationBroker broker, long millis)
+ {
+ int count = 0;
+ final long end = System.currentTimeMillis() + millis;
+ while (System.currentTimeMillis() < end)
+ {
+ try
+ {
+ if (broker.receive(false, false, true) instanceof TopologyMsg)
+ {
+ count++;
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ // nothing came within the timeout of the session
+ }
+ }
+ return count;
+ }
+
+ /** Waits until a replication server knows a replica of a peer, and the provided generation id of that peer. */
+ private void waitReplicaOfPeer(ReplicationServer replServer, DN baseDN, int peerId, long generationId)
+ throws Exception
+ {
+ newTimer().repeatUntilSuccess(() ->
+ {
+ final ReplicationServerHandler peer =
+ replServer.getReplicationServerDomain(baseDN).getConnectedRSs().get(peerId);
+ assertTrue(peer.hasRemoteLDAPServers(),
+ replServer.getMonitorInstanceName() + " knows a replica of RS(" + peerId + ")");
+ assertEquals(peer.getGenerationId(), generationId,
+ "generation id " + replServer.getMonitorInstanceName() + " has for RS(" + peerId + ")");
+ });
+ }
+
+ private void waitGenerationId(ReplicationServer replServer, DN baseDN, long generationId) throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertEquals(
+ replServer.getReplicationServerDomain(baseDN).getGenerationId(), generationId,
+ "generation id of " + replServer.getMonitorInstanceName()));
+ }
+
+ private void waitGenerationIdOfPeer(ReplicationServer replServer, DN baseDN, int peerId, long generationId)
+ throws Exception
+ {
+ newTimer().repeatUntilSuccess(() -> assertEquals(
+ replServer.getReplicationServerDomain(baseDN).getConnectedRSs().get(peerId).getGenerationId(),
+ generationId, "generation id " + replServer.getMonitorInstanceName() + " has for RS(" + peerId + ")"));
+ }
+
+ /** Returns the TopologyMsg a replication server with no replica sends its peers, with the provided generation id. */
+ private static TopologyMsg topologyMsgOf(ReplicationServer replServer, long generationId)
+ {
+ RSInfo rsInfo = new RSInfo(replServer.getServerId(), replServer.getServerURL(), generationId,
+ replServer.getGroupId(), replServer.getWeight());
+ return new TopologyMsg(Collections.<DSInfo> emptyList(), newArrayList(rsInfo));
+ }
+
@DataProvider
public Object[][] errorMsgIDsOfAPeer()
{
--
Gitblit v1.10.0