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

Valery Kharseko

yesterday 5055b6312ff61ef861be42bfb2827aa561bc1593

[#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)

Fixes #1235
Fixes #1236
Fixes #1240

## Problem

**#1235.** A replication server resets the generation ID of a domain
that had no change to -1 once it sees no directory server left in the
topology (`ReplicationServerDomain.resetGenerationIdIfPossible()`), and
tells its peers. A peer that still holds the generation ID - its own
directory server not gone yet, or its view of the others not updated yet
- took -1 in the TopologyMsg for a different generation ID in
`receiveTopoInfoFromRS`. It logged msgID 77 at WARNING and sent an
ErrorMsg back, which the first one logged as msgID 79 at ERROR, both
asking for a re-initialization. A stop of the whole combined topology in
run
[39909337627](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/39909337627)
logged it for `cn=schema`: two warnings on `cv-c1` and an error on each
of `cv-c0` and `cv-c3`. The handshake already leaves a peer at -1 alone
(`ReplicationServerHandler.checkGenerationId()`); only the TopologyMsg
path did not.

**#1236.** The replication server that reset takes a generation ID again
from the next TopologyMsg of a peer, through `setGenerationIdIfUnset`,
and told nobody. Its peers kept it at -1 and forwarded it no update
(`isDifferentGenerationId(rsHandler, updateMsg)`), until it sent a
TopologyMsg for some other reason. The handshake and
`changeGenerationId()` tell the peers; this path did not. Before #1235,
the warning 77 was the only trace of this in the logs.

**#1240.** The same silent take on the update path. A directory server
that connects with changes already made to a replication server without
a generation ID does not give it its generation ID at the handshake
(msgID 146). `put()` takes it from the first update of that directory
server, and told nobody either.

## Fix

In `ReplicationServerDomain`:
- `receiveTopoInfoFromRS` warns about a mismatch and sends the ErrorMsg
only for a peer which presents a generation ID (`> 0`), as
`checkGenerationId()` does. The generation ID of the peer is still
recorded, and updates are still not forwarded to a peer at -1: -1 is
also what `dsreplication pre-external-initialization` sets, so that the
replication servers take no change for the domain until it is
initialized again.
- `receiveTopoInfoFromRS` takes the generation ID only from a peer which
has directory servers, only while the replication server has none of its
own, and tells its peers when it takes it (`sendTopoInfoToAll()`).
- `put()` tells the peers when it takes the generation ID of an update.

Taking the generation ID only from a peer with directory servers is what
makes telling the peers safe. While those directory servers stay
connected, `resetGenerationIdIfPossible()` resets that generation ID
neither on the peer nor on the replication server that took it. A peer
without directory servers can reset the generation ID it presented as
soon as it receives a TopologyMsg. Two such replication servers whose
TopologyMsgs cross, one with a generation ID it has not saved and the
other at -1, would otherwise exchange it with no end: each one resets it
on the -1 of the other and takes it again from the next TopologyMsg of
the other, and each reset clears the changelog of the domain. A peer
without directory servers has no update of a directory server to forward
anyway, and the handshake still takes its generation ID.

A replication server with directory servers of its own takes no
generation ID from a TopologyMsg. With them connected,
`resetGenerationIdIfPossible()` never resets its generation ID, so it is
-1 only because a `ResetGenerationIdMsg` asked for it, such as the one
of `dsreplication pre-external-initialization`, or because they
connected with changes and gave none at the handshake (msgID 146), and
`put()` takes the one of their first update. A TopologyMsg that a peer
sent before it got the reset still presents the generation ID the reset
cleared, and it can be handled after the reset. build-maven
(ubuntu-latest, 21) of run
[40650839785](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/40650839785)
did that: the reader of the peer had read the TopologyMsg while the
reset held the lock of the domain. The replication server took the old
generation ID from it and told the peer, which took it in turn, and the
reset was undone on both for as long as their directory servers stayed
connected. Before this PR, that take was silent, and only the
replication server that handled the TopologyMsg went back to the old
generation ID. A replication server without directory servers which
takes such a generation ID resets it again on the TopologyMsg that the
peer sends once it is reset.

The take in `put()` cannot start such an exchange. It comes from a
directory server connected to this replication server, or from a peer
that forwards an update of its own directory server, which that peer
stored first. `resetGenerationIdIfPossible()` does not reset a
generation ID while a directory server is connected, nor once the
changelog has taken an update for it.

## Tests

In `ReplicationDomainTest`, with two replication servers in the JVM:
- `peerWithoutGenerationIdIsNotWarnedAbout` (#1235): the second
replication server sends the first one, which has a replica, a
TopologyMsg with -1 and then one with another generation ID. Neither
replication server logs the warning about -1, and the warning about the
other one still comes.
- `peerWhichTakesTheGenerationIdAgainIsForwardedUpdates` (#1236): the
second replication server resets its generation ID, then takes it again
from the TopologyMsg the first one sends when a second replica connects
to it. The first one records the generation ID again, and an update of
its replica reaches the second one.
- `peerWithoutReplicaGivesNoGenerationId` (#1236): a replication server
at -1 handles a TopologyMsg of a peer without a replica, with a
generation ID, and stays at -1. The TopologyMsg is handled in the test
thread, so that what the replication server tells the peer cannot come
back before the assertion.
- `peerWithReplicaGivesItsGenerationIdRightAfterAReset` (#1236): the
second replication server holds a generation ID it has not saved. One
TopologyMsg of the first replication server, which has a replica and
another generation ID, makes it reset its own and take that one in the
same call, and the first one records it.
- `peersWithoutReplicaEndWithoutGenerationId` (#1236): two replication
servers without a replica send each other the same unsaved generation
ID. Both end at -1 and stay there for two seconds.
- `topologyMsgSentBeforeAResetDoesNotUndoIt` (#1236): two replication
servers with a replica each are reset to -1 by a
ResetGenerationIdMsg(-1), as `dsreplication pre-external-initialization`
does. The first one then handles, in the test thread, the TopologyMsg
the second one built before the reset, with its replica and the old
generation ID, and stays at -1.
- `peerWithReplicaAndWithoutGenerationIdIsNotAnswered` (#1236): after
the same reset, a replication server without a replica handles, in the
test thread, a TopologyMsg of its peer with a replica and -1. In the
next two seconds, the replica of the peer receives no TopologyMsg, and
it is still connected: a closed session receives none.
- `generationIdTakenFromAnUpdateIsToldToThePeers` (#1240): a replica
with changes connects to the second replication server, which is at -1.
Its first update makes the second replication server take its generation
ID, the first one records it, and an update of the replica of the first
one reaches the second one.

The cases that change the generation ID of a replication server next to
a peer with a replica first wait until it knows that replica, so that no
TopologyMsg of the peer is left to cross the change.

## Verification

- **Red, run directly with TestNG on the cases above, with the class
under test replaced:**
- at b41b948 (the first commit of this PR):
`peerWithoutReplicaGivesNoGenerationId` (RS(61) at 6666 instead of -1,
three runs out of three) and
`generationIdTakenFromAnUpdateIsToldToThePeers` (RS(91) still at -1 for
RS(92)).
- taking the generation ID of a peer without a replica too:
`peerWithoutReplicaGivesNoGenerationId`, three runs out of three.
- taking nothing in the call that resets:
`peerWithReplicaGivesItsGenerationIdRightAfterAReset`, RS(72) at -1.
- `setGenerationIdIfUnset` taking -1 from a peer at -1:
`peerWithReplicaAndWithoutGenerationIdIsNotAnswered`, the replica
receives 1 TopologyMsg instead of none, two runs out of two.
- taking the generation ID while the replication server has directory
servers of its own, as the PR did up to 7f3af5eb70:
`topologyMsgSentBeforeAResetDoesNotUndoIt`, RS(81) at 5555 instead of
-1, two runs out of two.
- `put()` taking the generation ID silently:
`generationIdTakenFromAnUpdateIsToldToThePeers`.
  - Each mutant fails only its own case.
- **Without any fix:** `peerWithoutGenerationIdIsNotWarnedAbout` fails
on `msgID=77 ... RS(22) ... presented generation ID -1 ... RS(21) is
3333`, and `peerWhichTakesTheGenerationIdAgainIsForwardedUpdates` fails
with RS(41) still holding -1 for RS(42).
- **Green:** reactor `-Pprecommit verify -pl opendj-server-legacy -am
-Dit.test='org/opends/server/replication/**,org/opends/server/backends/ChangelogBackendTestCase'`
at this head, JDK 27: 4028 tests in 94 classes, no failure, none skipped
(1 h 11 min).
- **One-off probe, not kept as a case:** after the same reset, a
replication server without a replica handles a stale TopologyMsg of its
peer with a replica. It takes the old generation ID, the peer does not
take it back from it, and the next TopologyMsg of the peer, built after
the reset, makes it reset to -1 again.

## Not covered

- In a topology that mixes this version with an older one, a replication
server of this version still logs msgID 79 at ERROR when an older peer
warns it about its own -1. This lasts until every replication server
runs this version: the receiver cannot tell that warning apart from
others, because the generation IDs are only in the localized text.
- Updates published between the moment a replication server takes the
generation ID again and the moment its peers process its TopologyMsg are
still not forwarded to it. Closing this would mean reading the changelog
again for a peer replication server, which this PR does not do.
- A replication server at -1 next to a peer that has no directory server
but a saved generation ID no longer takes that generation ID from the
TopologyMsg of the peer, which says nothing of whether it is saved. It
takes it once a directory server connects to it or to that peer. A
directory server with changes that connects to it first gets msgID 146,
and the generation ID is taken from its first update (#1240). The same
holds for a replication server that took that generation ID at the
handshake and has not stored an update of that peer yet: the next
TopologyMsg of the peer resets it. If the directory server that connects
to it next has another generation ID, the replication server takes that
one from its first update, and it and the peer warn each other about the
mismatch (msgIDs 77 and 79) and exchange no update, where before this
change the directory server got BAD_GEN_ID. The mismatch is real either
way, since the peer holds changes of its own generation ID; only where
it is reported changes.
- A replication server whose directory servers connected with changes
while it was at -1 (msgID 146) no longer takes the generation ID of a
peer from its TopologyMsg. It takes the one of the first update of those
directory servers instead (#1240), which is the generation ID of their
data. Until that update, its peers forward it no update, as before this
PR, where it took the generation ID of the peer silently and the peers
still held -1 for it.
- Between `dsreplication pre-external-initialization` and
`post-external-initialization`, a replication server that takes the
generation ID of an update now tells its peers, so updates made in that
window are forwarded between replication servers that took the same
generation ID. This includes an update of a peer that was in flight when
the reset arrived: `put()` reads the generation ID of its source without
the lock of the domain, so it can take the generation ID from before the
reset, and a peer without directory servers of its own takes it in turn
from the TopologyMsg when this replication server has directory servers.
A directory server with an empty server state that connects to that peer
in this window then gets BAD_GEN_ID, where before this change the peer
stayed at -1 and took the generation ID of that directory server at the
handshake; the replication server that took the old one kept it to
itself. `post-external-initialization` sets the generation ID on every
replication server again.

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