[#1215] Check a replication server the topology already has against the con...

Valery Kharseko

yesterday d954c4c53f3d626f2a60c86eed653569dfd907fe

[#1215] Check a replication server the topology already has against the configured ones again, from the URL it is known with or the one it now comes with (#1218)

Fixes #1215

## Problem

A directory server left the replication server it was connected to every
two minutes, "because it is more suitable" to go to another one (msgID
188, `RS(X) was not configured locally on DS(N), but at least one other
RS was`), and connected straight back to it (msgID 62), for as long as
it ran. Each of those reconnections restarts the session, and under load
a reconnection is where #1206 lost changes.

`Topology.computeRSInfos()` decided whether a replication server is
locally configured only when the server first appeared in the topology
of the broker. A server the previous topology already had only went
through `ReplicationServerInfo.update()`, which keeps the flag and the
URL the server was first known with.

- A server a `TopologyMsg` named before the replication-server list of
the domain had it stayed not locally configured once the list got it.
That is the order of the background join of the Docker image, and of a
replication server added to a running topology.
- The restart of the session that the new list asks for keeps the
topology: `connectAsDataServer()` merges the infos of the phase-one
handshakes into it, and the periodic check of `receive()` then rejects
the server the broker is connected to by its first criterion,
`filterServersLocallyConfigured()`.
- The next connection evaluates the fresh phase-one infos, which take
every configured server as locally configured, and picks the same server
again.

The same loop follows, on master as well, when a configured server
announces a URL which is not the configured one (a port mapped by NAT or
by Docker) and the broker first hears of it from a `TopologyMsg`,
because it was down when the broker connected: the phase-one handshake
reaches it at the configured URL, but the topology keeps the announced
one.

## Change

`computeRSInfos()` now sets the flag of a server it updates again, in
the new `updateLocallyConfiguredFlag()`:

- the server is locally configured when the URL it is known with, or the
URL it comes with in the new topology, is one of the configured ones; in
the second case it is known with that URL from then on;
- a server which has the flag, but whose URL the configured ones no
longer have, is checked once more, resolving host names, with the URL it
is known with and then with the one it comes with, which is the URL a
new server is checked with; found so, it is known with the configured
URL from then on; any other loses the flag, and keeps the URL it is
known with;
- a new server is checked as before, by `setLocallyConfiguredFlag()`.

The lookup of a configured URL equivalent to a given one is now
`findEquivalentConfiguredUrl()`, which both `setLocallyConfiguredFlag()`
and the third check use.

The first two checks of a known server compare the strings only, and
resolve no host name. A server found locally configured is known with
the configured URL, and a phase-one handshake names a server with the
configured URL it reached it at. Resolving the hosts of every server on
every `TopologyMsg` would put a lookup per server and per configured URL
on the thread which receives the updates, and
`HostPort.isEquivalentTo()` returns false on an `UnknownHostException`:
a resolution failing for a moment would take the flag from the server
the broker is connected to, which is the reconnection this change is
about. A server with no URL (protocol version < 4) is checked apart: the
configured URLs are a sorted set, which throws on `contains(null)`.

The third check is there for an entry of the configuration written
another way, with an equivalent URL. The restart of the session which
that change asks for gives the flag back through the phase-one
handshake, but the restart is refused while a total update into this
replica runs, and in the request phase of the total update the periodic
check of `receive()` still runs: without the third check it would leave
the server the broker is connected to. The URL the server comes with is
looked up only when the one it is known with is not found: a server
known with the configured URL a phase-one handshake reached it at may
come with another one (a port mapped by NAT or by Docker), and the entry
may be written again as an equivalent of that one. Only a server with
the flag gets to the third check, and a server found so is then known
with the configured URL, so it costs at most two checks per change of
the configuration and per server which lost its URL. A resolution
failing at that moment takes the flag.

A change of the flag or of the URL of a known server is traced. The
previous topology holds the same `ReplicationServerInfo` object, so the
comparison of the two topologies in `topologyChange()` cannot show it.

Left as is: a server without the flag, known with or coming with a URL
which is only equivalent to a configured one, after resolution, gets the
flag at the next connection. A change of the list restarts the session,
and the phase-one handshake then gives the configured URL.

A server removed from the list was already left out: the restart builds
the topology from the phase-one infos, which only have the configured
servers, and `computeRSInfos()` keeps only the servers it is given. The
test of that case pins the same semantics in `Topology` itself.

## Tests

`ReplicationBrokerTest`, 13 new tests, 24 runs: the tests with a data
provider run through each of the two constructors of `Topology`.

| # | Test | What it pins |
|---|---|---|
| 1 |
`topologyShouldStampRSAddedToTheConfigurationAfterThePreviousTopology` |
A server a `TopologyMsg` named while the list lacked it is locally
configured, with the configured URL, once the list has it and the new
topology names it with that URL; the server the broker is connected to
still is. Red on master. |
| 2 |
`topologyShouldNotStampRSRemovedFromTheConfigurationAfterThePreviousTopology`
| A server the list no longer has is not locally configured. Red on
master. |
| 3 | `topologyShouldStampRSReachedAtAConfiguredURLItDoesNotAnnounce` |
`TopologyMsg` with the announced URL, then a phase-one handshake at the
configured URL, then a `TopologyMsg` with the announced URL again:
locally configured from the handshake on, with the configured URL. Red
on master. |
| 4 | `topologyShouldKeepStampedAConfiguredRSWithoutResolvingItsHost` |
A known server with a configured URL whose host cannot be resolved
(`rs.invalid`) stays locally configured, with that URL, when the new
topology names it with another URL. Skipped where the resolver answers
for `rs.invalid`. |
| 5 |
`topologyShouldStampRSAddedToTheConfigurationWithTheURLItAnnounces` |
The same as 1 with the URL the server was first known with, which is the
input of the background join of the Docker image. Red on master. |
| 6 | `bestServerCheckShouldNotRejectTheRSAddedToTheConfiguration` |
`computeBestReplicationServer()` does not reject the server of 5 as not
locally configured once the broker has reconnected to it. The other
server is not on this host, so the same-host filter keeps only this one:
the case pins the filter of the locally configured servers, not which
server the broker is connected to. Red on master. |
| 7 | `topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAnotherWay`
| A locally configured server whose entry is written `127.0.0.1:8989`
instead of `localhost:8989` keeps the flag, with the new URL. Red on
master. |
| 8 | `topologyShouldNotResolveTheURLAKnownRSComesWith` | A known server
without the flag which comes with a URL only equivalent to a configured
one stays without it. |
| 9 | `topologyShouldNotResolveTheURLAKnownRSIsKnownWith` | A known
server without the flag whose URL gets an equivalent in the list stays
without it. |
| 10 |
`topologyShouldStampRSAddedToTheConfigurationWithTheURLItIsKnownWith` |
A known server which comes with another URL, while the list now has the
URL it is known with, is locally configured, with that URL. IP literals
with distinct ports, so that no resolver can make it pass without the
first check. Red on master. |
| 11 | `topologyShouldNotStampKnownRSWithoutURL` | A known server with
no URL is not locally configured, and the check does not ask a sorted
set of configured URLs, as the configuration is, for a null one. |
| 12 |
`topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAsTheURLItComesWith`
| A locally configured server, known as `localhost:28989` from the
phase-one handshake, comes with `localhost:8989`, and the entry is
written `127.0.0.1:8989`: it keeps the flag, with the new URL. Red on
master. |
| 13 | `topologyShouldNotStampRSWhoseConfiguredURLNamesNeitherOfItsURLs`
| The same server, with the entry replaced with `127.0.0.1:18989`,
equivalent to neither of its URLs: it loses the flag, and keeps the URL
it is known with. Red on master. |

Mutants of `ReplicationBroker`, each run against the class:

| Mutant | Caught by |
|---|---|
| a known server checked again as a new one, resolving host names | 9,
12 (both constructors), 4 (both constructors, where `rs.invalid` does
not resolve), 3 |
| the flag of a known server can only be raised | 2, 7, 12, 13 (both
constructors) |
| the flag set without the URL it was found with | 1 (both
constructors), 3 |
| only the URL the server is known with counts | 1 (both constructors),
3 |
| no first check: only the URL the server comes with counts | 10 (both
constructors), 4 (both constructors, where `rs.invalid` does not
resolve) |
| the URL the server is known with keeps the flag it had | 5, 10 (both
constructors), 6 |
| a server with the flag which lost its URL is not checked again | 7, 12
(both constructors) |
| a server without the flag which lost its URL is checked again as well
| 8, 9 (both constructors) |
| the URL the server comes with matched by resolving host names | 7, 8,
9, 12 (both constructors) |
| the third check looks up only the URL the server is known with | 12
(both constructors) |
| a server which loses the flag takes the URL it comes with | 13 (both
constructors) |
| a server with no URL looked up in the configured URLs | 11 (both
constructors) |

Not pinned: the trace. Dropping it leaves the class green; it is a
diagnostic, and its format and its six arguments were checked by
reading.

## Verification

- `ReplicationBrokerTest`, run directly with TestNG, JDK 21: 37 tests, 0
failures; 16 of them fail on master. On a resolver which answers for
`rs.invalid` (`-Djdk.net.hosts.file`), test 4 is skipped in both runs
and the other 35 pass.
- `org/opends/server/replication/**` in the reactor (`-pl
opendj-server-legacy -am`), on master b8a4d35f03, JDK 27, macOS: 3946
tests, 0 failures, 0 errors, 0 skipped.

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java

@@ -3126,6 +3126,8 @@
        {
          // Update the existing info for the replication server
          rsInfo.update(newRSInfo, connectedDSs);
          updateLocallyConfiguredFlag(rsInfo, newRSInfo.getServerUrl(),
              configuredReplicationServerUrls, dsServerId);
        }
      }

@@ -3184,19 +3186,132 @@
        rsInfo.setLocallyConfigured(false);
        return;
      }
      for (String serverUrl : configuredReplicationServerUrls)
      final String serverUrl =
          findEquivalentConfiguredUrl(rsUrl, configuredReplicationServerUrls);
      if (serverUrl != null)
      {
        if (isSameReplicationServerUrl(serverUrl, rsUrl))
        {
          // This RS is locally configured, mark this
          rsInfo.setLocallyConfigured(true);
          rsInfo.setServerURL(serverUrl);
          return;
        }
        // This RS is locally configured, mark this
        rsInfo.setLocallyConfigured(true);
        rsInfo.setServerURL(serverUrl);
        return;
      }
      rsInfo.setLocallyConfigured(false);
    }

    /**
     * Sets the locally configured flag for a ReplicationServerInfo object the
     * previous topology already had: the configured replication servers may
     * have changed since its flag was set.
     * <p>
     * The server is locally configured when the URL it is known with, or the
     * one it now comes with, is a configured one. This needs no host name
     * resolution, unlike setLocallyConfiguredFlag(): a server found
     * locally configured is known with the configured URL, and a phase-one
     * handshake names a server with the configured URL it reached it at. So a
     * resolution failing for a moment cannot take the flag from a server known
     * with a configured URL.
     * <p>
     * A server which has the flag, but whose URL the configured ones no longer
     * have, is checked once more, resolving host names, with the URL it is known
     * with and then with the one it comes with, which is the URL a new server is
     * checked with: the configured URL may only have been written another way,
     * and the restart of the session which a change of the configuration asks
     * for is refused while a total update into this replica runs. A server found
     * so is known with the configured URL from then on; any other loses the
     * flag, keeps the URL it is known with, and is not resolved again. A
     * resolution failing at that moment takes the flag.
     *
     * @param rsInfo
     *          the Replication server to check and update
     * @param newServerUrl
     *          the URL the server comes with in the new topology
     * @param configuredReplicationServerUrls
     *          the configured replication server URLs
     * @param dsServerId
     *          the server id of the local DS, for the trace
     */
    private void updateLocallyConfiguredFlag(ReplicationServerInfo rsInfo,
        String newServerUrl, Set<String> configuredReplicationServerUrls,
        int dsServerId)
    {
      final boolean wasLocallyConfigured = rsInfo.isLocallyConfigured();
      final String knownServerUrl = rsInfo.getServerURL();
      if (isConfigured(knownServerUrl, configuredReplicationServerUrls))
      {
        rsInfo.setLocallyConfigured(true);
      }
      else if (isConfigured(newServerUrl, configuredReplicationServerUrls))
      {
        rsInfo.setLocallyConfigured(true);
        rsInfo.setServerURL(newServerUrl);
      }
      else if (wasLocallyConfigured)
      {
        String configuredUrl = findEquivalentConfiguredUrl(knownServerUrl,
            configuredReplicationServerUrls);
        if (configuredUrl == null
            && !Objects.equals(newServerUrl, knownServerUrl))
        {
          configuredUrl = findEquivalentConfiguredUrl(newServerUrl,
              configuredReplicationServerUrls);
        }
        rsInfo.setLocallyConfigured(configuredUrl != null);
        if (configuredUrl != null)
        {
          rsInfo.setServerURL(configuredUrl);
        }
      }
      else
      {
        rsInfo.setLocallyConfigured(false);
      }

      // The previous topology holds the same object, so its comparison with the
      // new one cannot show this change
      if (logger.isTraceEnabled()
          && (wasLocallyConfigured != rsInfo.isLocallyConfigured()
              || !Objects.equals(knownServerUrl, rsInfo.getServerURL())))
      {
        logger.trace("RS(%d) on DS(%d): locally configured %s -> %s, URL %s -> %s",
            rsInfo.getServerId(), dsServerId, wasLocallyConfigured,
            rsInfo.isLocallyConfigured(), knownServerUrl, rsInfo.getServerURL());
      }
    }

    private boolean isConfigured(String rsUrl,
        Set<String> configuredReplicationServerUrls)
    {
      // A server with no URL (protocol version < 4) cannot be connected to
      return rsUrl != null && configuredReplicationServerUrls.contains(rsUrl);
    }

    /**
     * Returns the configured replication server URL which names the same server
     * as the passed URL, resolving host names.
     *
     * @param rsUrl
     *          the URL of the server, or null if it has none
     * @param configuredReplicationServerUrls
     *          the configured replication server URLs
     * @return the configured URL which names the same server, or null if none
     *         does or if the server has no URL
     */
    private String findEquivalentConfiguredUrl(String rsUrl,
        Set<String> configuredReplicationServerUrls)
    {
      if (rsUrl != null)
      {
        for (String serverUrl : configuredReplicationServerUrls)
        {
          if (isSameReplicationServerUrl(serverUrl, rsUrl))
          {
            return serverUrl;
          }
        }
      }
      return null;
    }

    @Override
    public boolean equals(Object obj)
    {

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationBrokerTest.java

@@ -12,24 +12,31 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

import org.opends.server.DirectoryServerTestCase;
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.protocol.TopologyMsg;
import org.opends.server.replication.service.ReplicationBroker.RSEvaluations;
import org.opends.server.replication.service.ReplicationBroker.ReplicationServerInfo;
import org.opends.server.replication.service.ReplicationBroker.Topology;
import org.testng.Assert;
import org.testng.SkipException;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import static java.util.Collections.*;

import static org.assertj.core.api.Assertions.*;
import static org.opends.server.replication.service.ReplicationBroker.computeBestReplicationServer;
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

@@ -254,4 +261,347 @@
    assertThat(currentRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.isLocallyConfigured()).isFalse();
  }

  /**
   * A topology names a replication server before the configuration of the domain lists it: the next
   * topology, built once the configuration lists it, must stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationAfterThePreviousTopology(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo configuredRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), configuredRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo anotherRS = find(after.rsInfos, configuredRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:1389");
    assertThat(find(after.rsInfos, CURRENT_RS.toRSInfo()).isLocallyConfigured()).isTrue();
  }

  /**
   * A configured replication server announces in the topology messages a URL other than the
   * configured one (a port mapped by NAT or by Docker), and the broker hears of it first that way.
   * Once a phase-one handshake has reached it at the configured URL, it is locally configured, and
   * stays so when the next topology message names it with the announced URL again.
   */
  @Test
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSReachedAtAConfiguredURLItDoesNotAnnounce()
  {
    final Set<String> configured = newHashSet(CURRENT_RS.getServerURL(), "localhost:28989");
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology heard = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    assertThat(find(heard.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, configured, heard.rsInfos);
    ReplicationServerInfo anotherRS = find(connected.rsInfos, reachedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:28989");

    final Topology next = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, configured, connected.rsInfos);
    anotherRS = find(next.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:28989");
  }

  /**
   * A replication server the previous topology stamped as locally configured stays so as long as the
   * configuration lists the URL it is known with, even when the topology names it with another URL (a port
   * mapped by NAT or by Docker) and its host name cannot be resolved at that moment: losing the flag
   * would make the broker leave the server it is connected to.
   * <p>
   * The case shows that the host is not resolved only where resolving it fails: a resolver which
   * answers for any name (the fake addresses of a proxy or VPN DNS) skips it.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedAConfiguredRSWithoutResolvingItsHost(TopologyCtorToUse toUse)
  {
    try
    {
      final InetAddress[] answer = InetAddress.getAllByName("rs.invalid");
      throw new SkipException("The resolver answers for rs.invalid with " + Arrays.toString(answer)
          + ": this case could not tell a resolved host from one that is not");
    }
    catch (UnknownHostException expected)
    {
      // The premise of this case: resolving the host fails
    }
    final String unresolvableURL = "rs.invalid:1389";
    final RSInfo connectedRS = new RSInfo(ANOTHER_RS_ID, unresolvableURL, 0, (byte) 0, 0);
    final Map<Integer, ReplicationServerInfo> previousRSs =
        newMap(new ReplicationServerInfo(connectedRS, newHashSet(CURRENT_DS.getDsId())));

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "rs.invalid:8989", 0, (byte) 0, 0);
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(unresolvableURL), previousRSs);
    assertInvariants(topo);
    final ReplicationServerInfo anotherRS = find(topo.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo(unresolvableURL);
  }

  /**
   * A replication server is removed from the configuration of the domain: the next topology must no
   * longer stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampRSRemovedFromTheConfigurationAfterThePreviousTopology(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isTrue();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A topology names a replication server before the configuration of the domain lists it, with the URL
   * the configuration then lists it with: the next topology must stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationWithTheURLItAnnounces(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isTrue();
  }

  /**
   * The broker has reconnected to a replication server a topology named before the configuration of
   * the domain listed it, and the topology of the new session has the infos of the phase-one handshakes:
   * the periodic check of the best replication server must not reject it as a server which is not
   * locally configured.
   * <p>
   * The other server is not on this host, so the same-host filter keeps only this one, and the check
   * returns it before it reads which server the broker is connected to: the case pins the filter of the
   * locally configured servers only.
   */
  @Test
  @SuppressWarnings("unchecked")
  public void bestServerCheckShouldNotRejectTheRSAddedToTheConfiguration()
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);

    final Topology reconnected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    final RSEvaluations evals = computeBestReplicationServer(false, ANOTHER_RS_ID, new ServerState(),
        reconnected.rsInfos, CURRENT_DS.getDsId(), (byte) 0, 0);
    assertThat(evals.getBestRS()).isNotNull();
    assertThat(evals.getBestRS().getServerId()).as(evals.toString()).isEqualTo(ANOTHER_RS_ID);
  }

  /**
   * The entry of the configuration which lists a replication server stamped as locally configured is
   * written another way, with an equivalent URL: the server must keep the flag, with the configured URL,
   * even when the restart of the session this asks for is refused (while a total update runs).
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAnotherWay(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:8989"), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isTrue();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:8989"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo anotherRS = find(after.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("127.0.0.1:8989");
  }

  /**
   * A known replication server which is not locally configured comes with a URL which is only an
   * equivalent of a configured one: the topology message must not resolve it, so the server gets the flag
   * at the next connection only.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotResolveTheURLAKnownRSComesWith(TopologyCtorToUse toUse)
  {
    final Set<String> configured = newHashSet(CURRENT_RS.getServerURL(), "localhost:28989");
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo equivalentRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:28989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), equivalentRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, equivalentRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A known replication server which is not locally configured gets an equivalent of its URL in the
   * configuration of the domain, written another way: the next topology must not resolve it either, so
   * the server gets the flag at the next connection only.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotResolveTheURLAKnownRSIsKnownWith(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:28989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A known replication server comes with another URL in the next topology (a mapped port) while the
   * configuration of the domain now lists the URL it is known with: it is locally configured, without
   * resolving any host name.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationWithTheURLItIsKnownWith(TopologyCtorToUse toUse)
  {
    final RSInfo knownRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.2:28989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), knownRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, knownRS).isLocallyConfigured()).isFalse();

    final RSInfo mappedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.3:28989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), mappedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.2:28989"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, mappedRS);
    assertThat(rs.isLocallyConfigured()).isTrue();
    assertThat(rs.getServerURL()).isEqualTo("127.0.0.2:28989");
  }

  /**
   * A known replication server with no URL (protocol version below 4) is not locally configured, and the
   * check must not ask the configured URLs for a null one: they are a sorted set, as in the configuration,
   * whose contains(null) throws.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampKnownRSWithoutURL(TopologyCtorToUse toUse)
  {
    // sorted, as the configuration is: contains(null) throws
    final Set<String> configured = new TreeSet<>(Arrays.asList(CURRENT_RS.getServerURL(), "localhost:8989"));
    final RSInfo noUrlRS = new RSInfo(ANOTHER_RS_ID, null, 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), noUrlRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), noUrlRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, before.rsInfos);
    assertThat(find(after.rsInfos, noUrlRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A locally configured replication server, known with the configured URL a phase-one handshake reached
   * it at, comes with another URL (a port mapped by NAT or by Docker), and the entry of the configuration
   * is written another way, as an equivalent of the URL it comes with: the server must keep the flag,
   * with the configured URL.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAsTheURLItComesWith(TopologyCtorToUse toUse)
  {
    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), EMPTY_MAP);
    assertThat(find(connected.rsInfos, reachedRS).isLocallyConfigured()).isTrue();

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:8989"), connected.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, announcedRS);
    assertThat(rs.isLocallyConfigured()).isTrue();
    assertThat(rs.getServerURL()).isEqualTo("127.0.0.1:8989");
  }

  /**
   * A locally configured replication server, known with the configured URL a phase-one handshake reached
   * it at, comes with another URL, and the entry of the configuration is replaced with a URL equivalent to
   * neither: the server loses the flag, and keeps the URL it is known with.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampRSWhoseConfiguredURLNamesNeitherOfItsURLs(TopologyCtorToUse toUse)
  {
    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), EMPTY_MAP);
    assertThat(find(connected.rsInfos, reachedRS).isLocallyConfigured()).isTrue();

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:18989"), connected.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, announcedRS);
    assertThat(rs.isLocallyConfigured()).isFalse();
    assertThat(rs.getServerURL()).isEqualTo("localhost:28989");
  }
}