Replication port stolen by TCP self-connect: RS silently starts without listener (testSafeDataLevelOne CI flake)

[vharseko]

Symptom

AssuredReplicationServerTest.testSafeDataLevelOne fails intermittently on CI
during test setup, e.g. the ubuntu-latest/25 leg of PR #727's run
(unrelated to that PR):

AssuredReplicationServerTest.testSafeDataLevelOne:1156
  ->createFakeReplicationDomain:321 expected [true] but found [false]

Line 321 is assertTrue(fakeReplicationDomain.isConnected()). The server log
shows the actual cause right before it:

msgID=6 Replication Server failed to start : could not bind to the listen port : 65531.
Error : Address already in use
msgID=208 Directory server DS(1) was unable to connect to any replication servers for domain "o=test"

The test is a data provider with 28 iterations; each iteration recreates the
real RS on the same fixed port about a second after the previous iteration
shut it down.

Root cause: TCP self-connect steals the replication port

Ruled out first: TIME_WAIT does not explain it — the JDK enables
SO_REUSEADDR on ServerSocket by default on POSIX platforms, so binding
over TIME_WAIT remnants already works. The port must have been held by a
live socket.

The thief is the well-known TCP simultaneous open self-connect
(JDK-8067207; same bug class as ZOOKEEPER-2101, KAFKA-1836):

1. Test listen ports come from TestCaseUtils.findFreePorts() = bind(0),
   i.e. they are drawn from the OS ephemeral port range (proven by the
   fact the OS assigned 65531 in the first place).
2. When an RS is shut down between test iterations, its former peers
   (ReplicationBroker of fake domains, other RSs) keep reconnecting to
   that port in a tight retry loop while nothing is listening.
3. For each connect() the kernel assigns a local ephemeral port. Sooner or
   later it picks the destination port itself — on Linux the connect then
   succeeds via TCP simultaneous open, and the socket is connected to
   itself, occupying the very port the next RS is about to bind.
4. The next new ReplicationServer(...) fails to bind —
   initialize() only logs ERR_COULD_NOT_BIND_CHANGELOG and returns
   normally, so the RS object exists without any listener and the fake
   domain's isConnected() assertion fails.

macOS/BSD stacks refuse self-connects (EINVAL), which is consistent with
the flake being ubuntu-only. The per-attempt probability is ~1/ephemeral-range,
and the reconnect storm produces hundreds of attempts per run — matching the
observed low-but-recurring frequency.

Fix

Detect and reject self-connected sockets in both outgoing replication connect
paths, the same way Kafka/ZooKeeper fixed the identical bug:

• StaticUtils.isSelfConnection(Socket) — local endpoint == remote endpoint;
• ReplicationBroker.performPhaseOneHandshake() — throw ConnectException
  (existing handling closes the socket, releasing the port, and retries);
• ReplicationServer.connect() — same, the existing failure path also
  blacklists the address for a few connect iterations.

A legitimate established connection can never have identical local and remote
endpoints, so the check cannot produce false positives.

Follow-up worth discussing separately

1. ReplicationServer.initialize() swallows the bind IOException (log
   only): the constructor "succeeds" with no listen thread, nothing surfaces
   to the caller, and peers just fail to connect. Fail-fast (or bind retry)
   would have made this flake a one-line diagnosis instead of a stochastic
   isConnected() failure.
2. StaticUtils.isAddressInUse() probes wildcard addresses with a "dummy
   connect" — on Linux that probe itself can self-connect to an unbound port
   and wrongly report the port as in use.