[#1235] [#1236] [#1240] Replication: no mismatch warning for a peer without...

Valery Kharseko

yesterday 5055b6312ff61ef861be42bfb2827aa561bc1593

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


 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()
  {