[#1202] [#1209] [#1222] Give the replication server a bounded time to give ...

Valery Kharseko

yesterday 77ad03e14cc1bbc5471f332ba73430d4c472b366

[#1202] [#1209] [#1222] Give the replication server a bounded time to give credit and to read once the broker is being stopped, and write nothing after a message given up (#1210)

Fixes #1202
Fixes #1209
Fixes #1222

## Summary

`ReplicationBroker.stop()` could not end while its replication server
kept the session up and took no update from it, and with it hung
`ReplicationDomain.disableService()`, which calls `stop()` holding
`serviceStateLock` and `sessionLock` - the shutdown of the domain, an
import, a restore, a session restart. It hung in two ways: waiting for
window credit the replication server did not give (#1202), and, once the
socket buffers were full, in a write to the session which the
replication server did not read (#1209) - with the default window of
100000 updates, the buffers fill long before the window closes. The stop
now gives the replication server a bounded time to give credit, and as
long again to read what was written, after which it closes the session
as a failed one; once a publish gave its message up for the stop, the
broker writes nothing more. The same write held the session with no stop
at all once the replication server was gone without a reset (#1222): the
heartbeat monitor, which then closes the session, waited for it as well,
until TCP gave up retransmitting. It now closes such a session as a
failed one too.

## Root cause

- The one flag which ends the wait of a publish for window credit is
`shutdown` (`while (!done && !shutdown)` in
`ReplicationBroker.publish()`), and `stop()` set it only after
`domain.publishReplicaOfflineMsg()` returned.
- `ReplicaOfflineMsg` extends `UpdateMsg`: it takes a permit of the send
window like a change. With the window spent, the stop waited for the
credit of its own replica offline message even with no local change in
flight - wider than the case #1202 describes.
- In the case of #1202, a local change waits for credit inside
`PendingChanges.commitAndPushCommittedChanges()`, under the monitor
`PendingChanges.putReplicaOfflineMsg()` needs: the stop waits for that
change, which waits for credit.
- Credit comes only as the replication server takes the updates of the
session, while its heartbeats come from a thread of their own and keep
the session up. Nothing else ended the wait.
- A directory server writes its session on the thread which publishes -
its `Session` runs no publisher thread - and a local change is published
under the lock of the pending changes. A write which the replication
server does not read blocks in the socket once the buffers are full,
holding that lock, `connectPhaseLock` and the `publishLock` of the
session (#1209). A socket write has no timeout, and `Session.close()` of
a session with no error takes `publishLock` first: nothing on the
directory server ended that write.
- The heartbeat monitor, which closes the session once the heartbeats of
the replication server stop, closed it with `Session.close()` too
(#1222). A replication server which is gone without a reset acknowledges
nothing: the writes fill the buffers, the next one blocks, and the close
waited for it until TCP gave up retransmitting - while the listener
stayed in a read which neither data nor a reset ends, and the local
changes waited for the lock of the pending changes.

## Change

`ReplicationBroker`:

- `stop()` records the stop in progress, with a deadline, before it
publishes the replica offline message: a `StopInProgress` in the
`volatile` field `stopping`, which `start()` clears to `null`. The
deadline is the connection timeout of the replication provider
(`MultimasterReplication.getConnectionTimeoutMS()`, 5 seconds by
default): the replication server gets as long to give credit as it gets
to accept a connection. One which is only slow still takes the changes
and the replica offline message, as before. A connection timeout of 0,
which sets no limit to a connection or to an SSL negotiation, sets none
to the stop either: the stop then waits as it did before.
- A stop which overlaps one in progress keeps that one, its deadlines
and what it gave up, rather than replacing it.
- A publish which finds no credit once the deadline is past gives its
message up and marks the stop (`StopInProgress.gaveUp`), which ends the
publish loop for every later message as well, until `start()` clears the
stop. Before it marks the stop it reads it again: if `start()` ran
since, the loop head decides. What is given up:
- a local change, which the next session republishes from the historical
information of its entry - or a local delete, one the recovery publishes
included, from the deletes the domain keeps aside in memory, which a
restart of the server loses;
- or the replica offline message, which
`PendingChanges.putReplicaOfflineMsg()` then reports as not sent, its
announcement withdrawn.
- Nothing is written by `publish()` after a message given up: the
replica offline message carries the newest CSN of the replica, and
written after a change given up it would tell the replication server
that nothing older is to come.
- The give-up is marked on the stop, not on the broker. A publish which
found the stop past its deadline may mark it only after `start()`
cleared it - the read and the write are not atomic, and no lock is taken
for them - and it then marks a stop which is over, rather than the
broker started again, which would refuse every publish until its next
stop.
- `stop()` starts a watchdog for the stop, unless the connection timeout
is 0 (`endWhatStillHoldsTheStop()`). If the stop is still held when it
is past its credit deadline by as long again - twice the connection
timeout from its start - what holds it is no wait for credit, which
gives up by itself at the deadline, but a write the replication server
does not read, or a lock such a write holds. The watchdog then marks the
give-up of the stop, logs a warning which names the replication server,
and closes the session as a failed one (`Session.abort()`): the write
fails, and the publish which made it returns rather than retries. A stop
which waits for credit and gives up still ends its session in order,
with its `StopMsg`.
- The watchdog acts under the monitor of the stop, and `stop()` ends the
stop under `startStopLock` once it is done: the watchdog of a stop never
closes a session of a broker started again.
- A publish reads the give-up again under `connectPhaseLock`, before it
writes a message and before it writes a window probe. A publish which
passed the loop head before the give-up may wait for that lock behind a
connection of the listener - one which the replication server ending the
session in the stop set off, and which takes the connection timeout for
each replication server that answers nothing - past the deadline of the
watchdog, which then finds no session to close. Without that read the
publish wrote to the session the connection opened, and a replication
server which did not read it would hold the stop with nothing left to
end it. The watchdog marks the give-up before it reads the session, and
a connection sets its session before it releases the lock: either the
watchdog closes that session, or the publish finds the mark.
- Once the stop gave up, the broker opens no session again while the
stop runs (`reStart()`): the listener, which the close of the session
wakes, waits for the end of the stop rather than connects. A session
opened then would carry nothing but its change time heartbeats - which
tell the replication server that nothing older is to come from this
replica, while it still has the changes given up - and the stop would
wait for the connection. A connection already under way when the stop
gives up still opens its session: `publish()` writes nothing to it, and
the stop closes it as it ends.
- `stop()` stops the broker in a `finally`: a failure of the replica
offline message no longer leaves a stopping broker running, which would
give up every publish that waits for credit. The watchdog is started
inside the same `try`: a thread which cannot be started leaves no
stopping broker running either.

`Session.abort()` (new) records the error, marks the close begun, closes
the sockets, and then closes the session, which takes the branch of a
failed session: it writes nothing more - neither what the publisher
thread left queued nor the `StopMsg` - and takes no lock. The sockets
closed first end a write blocked in them, whichever thread makes it -
the publisher thread, which any close joins, included - and release a
close which had begun before, with no error, and waits for the lock of
the writes. The close is marked begun before them, so that the listener
whose read their close ends finds it begun, and does not report it as a
failure of the replication server.

`HeartbeatMonitor` closes a session whose heartbeats stopped with
`Session.abort()` (#1222): the blocked write and the read of the
listener fail, and the reconnection proceeds as for any session which
failed; no `StopMsg` is written, which a replication server whose
heartbeats stopped is not expected to read. A monitor stopped with its
broker still closes the session as before.

The javadoc of `ReplicationDomain.publish(UpdateMsg)` and the trace of
`LDAPReplicationDomain.publishReplicaOfflineMsg()` name the new reasons
a message is not published, and the javadoc of
`DSRSShutdownSync.replicaOfflineMsgNotSent()` counts the wait of a
stopping broker in the window between an announcement and its
withdrawal. The synopsis of `connection-timeout`
(`ReplicationSynchronizationProviderConfiguration.xml`) says that it
also bounds the wait of a stopping domain for credit, and its
description says when a domain is stopped, what raising or lowering the
value does to such a stop, that a replication server which does not read
what was written gets as long again before the session is closed, and
that 0 sets no limit.

## Not changed

- A publish held up by an `IOException` still retries until the broker
is connected again or finds no replication server to connect to - unless
the stop gave up.
- A recovery publish given up is refused to the recovery, as one refused
once `shutdown` is set (#1204): the round ends, and the changes from it
- a delete included, which stays aside - are left to the recovery of the
next session.
- An assured change given up for the stop still waits for its
acknowledgment until the assured timeout, and is counted as sent and
timed out. Ending that wait early needs a mark of its own in the assured
machinery: a CSN taken out of the messages which wait for an
acknowledgment reads as acknowledged.
- A change given up comes back with the recovery of the next session,
which runs in a thread of its own, and the change time heartbeats of
that session do not wait for it. When the recovery takes longer than a
heartbeat interval, the replication server takes a heartbeat as "nothing
older is to come" from this replica, and the change gets its change
number after newer changes of other replicas. Nothing is lost; any
change a recovery publishes does the same on master.
- The change time heartbeats, and the window and status messages written
straight to the session, are not held back after a give-up: only
`publish()` reads the mark, until the session is closed. That holds for
the session a connection already under way at the give-up opens too,
until the stop closes it as it ends.
- A replication server which drops the session after a give-up for
credit is reported (`WARN_REPLICATION_SERVER_BADLY_DISCONNECTED`), as
with no stop at all: the give-up closes nothing. Only a close this
server makes - by the watchdog of a stop, or by the heartbeat monitor -
is not, as the close is marked begun before the sockets are closed.
- A total update this domain is sending ends with an error once the stop
gives up, as every message of it is refused - as it does on master as
soon as a stop with the window open sets `shutdown`. Only the stop which
waited for credit for good let it go on.
- A session the watchdog or the heartbeat monitor closes ends with no
`StopMsg`: the replication server sees the connection lost.
- A reconnection of the listener to a better replication server
(`reStart()` from `receive()`) still closes the current session with no
error, and waits for a write blocked in it.
- With the window open, a stop still waits for the end of a round #1204
holds the local changes back for, as #1202 records.

## Tests

`ReplicationBrokerStopTest` (new) connects a domain to a fake
replication server which keeps the session up and reads all of it, but
opens a window of one update - or of the default 100000, in the cases of
a write it does not read - and gives no credit after it unless the test
asks it to. Like a replication server, it reports as its state the
changes it received; it serves each connection on a thread of its own,
and stops reading its sessions when the test asks it to. It sends the
heartbeats the domain asks for, and can go away altogether: no reading,
no heartbeat, no answer to a connection. It ends a session in order or
drops it when the test asks it to. In the case of a connection under
way, more replication servers take each connection and, once told to,
answer nothing on it, as one which hangs does: a connection to each
waits for the connection timeout in its SSL negotiation. Once the domain
closed its session while the server read nothing, the server ends that
session with a heartbeat, which the closed end answers with a reset:
what the closed socket left queued, its end included, comes only once
its TCP finds the window open again, and after a stall of 14 seconds its
next window probe is 13 seconds away on Linux - the CI run of round 2
failed `aDisableEndsAWrite…` waiting for that end. A publish which waits
for credit shows as a `WindowProbeMsg` on that server; "the stop waits
for the lock" is the disabling thread `BLOCKED` in
`PendingChanges.putReplicaOfflineMsg()`; a write it does not read is a
writer which stays in `Session.send()` while the number of its changes
stands still for two seconds in a row - half a second could be a pause
of the collector, as a run after cases which allocate a lot showed;
"reported as a close by the replication server" is a
`WARN_REPLICATION_SERVER_BADLY_DISCONNECTED` in the error log; "the
watchdog acted" is the end of its thread. The class runs with a
connection timeout of 2 seconds instead of the 30 seconds
`config-changes.ldif` gives the suite, and a case which needs another
one sets it for its own run.

| Case | What it pins | master | round 1 |
|---|---|---|---|
| `aDisableEndsWhenTheReplicationServerGivesNoCredit` | Once a change
spent the window, disabling the domain ends, the replica offline message
not written - no sooner than the connection timeout of 6 seconds the
case sets, and within 22 seconds; the announcement of the message is
withdrawn, and the session ends in order, with a `StopMsg`. | red |
green |
| `withNoConnectionTimeoutAStopWaitsForTheCredit` | With a connection
timeout of 0, the replica offline message still waits for credit past
its first wait and past 7 seconds, beyond the shipped default, and goes
out once the credit comes. | green | green |
| `aDisableEndsWhileALocalChangeWaitsForCredit` | Disabling while a
second change waits for credit under the lock ends, and the change ends
with it. | red | green |
| `aStopPublishesWhatTheReplicationServerTakesInTime` | Credit given a
second into the stop gets that change out, and then the replica offline
message, last. | green | green |
| `nothingIsWrittenAfterAChangeTheStopGaveUp` | A change given up is
followed by nothing, even with credit offered right after it; a broker
started again republishes it with the recovery of its session, and then
writes again. | red | green |
| `aSessionRestartRepublishesTheChangeTheStopGaveUp` | A restart of the
session (`restartService()`) gives the change up as a disable does, and
the recovery of the next session republishes it. Unlike `enable()`,
which loads the state again and repairs it from `ds-sync-hist`, a
restart keeps the state in memory: the change is found missing only
because `ReplicationDomain.publish()` put it in the state although the
broker did not write it. | red | green |
| `aStopWritesNothingAfterAGiveUpEvenWithCredit` | A stop which gave a
change up refuses the next one with the window open again, and an
`EntryMsg` and a `DoneMsg`, which need no credit: no race with the 500
ms wait for credit. | red | green |
| `aStopWhichOverlapsOneInProgressKeepsItsGiveUp` | After a change given
up, the first stop writes nothing more, with the window open again,
while a second stop runs beside it. | red | red |
| `aBrokerStartedAgainWaitsForCredit` | A stop with the window open
writes the replica offline message; a broker started again waits for
credit past twice the connection timeout of that stop, and keeps its
session. | green | green |
| `aStopWhichCannotPublishTheOfflineMessageStillStopsTheBroker` | A stop
whose replica offline message fails rethrows the failure and still stops
the broker. | red | green |
| `aDisableEndsAWriteToAReplicationServerWhichReadsNothing` | A local
change blocked in its write under the lock of the pending changes, with
credit to spare: disabling ends, no sooner than twice the connection
timeout of 6 seconds and sooner than two and a half times it; nothing is
written after the change, no session is opened again during the stop,
nothing reports the close as one by the replication server, and the
recovery of the session after `enable()` republishes the change. | red |
red |
| `aStopEndsItsOwnWriteAndOpensNoSessionAgain` | A write of the stop
itself, with no lock of the pending changes in the way, ends the same
way, no sooner than twice the connection timeout; the listener the close
wakes waits for the end of the stop rather than connects, nothing
reports the close as one by the replication server, and nothing is
written after the write given up. | red | red |
| `theHeartbeatMonitorEndsAWriteToAReplicationServerWhichIsGone` | A
local change blocked in its write to a replication server which then
reads nothing, sends no heartbeat and answers no connection: the write
ends, no sooner than two heartbeat intervals, nothing reports the close
as one by the replication server, and once the server is back the domain
connects again and its change reaches it. | red | red |
| `aConnectionUnderWayWhenTheStopGivesUpGetsNothingWritten` | The
replication server ends the session in the stop, and four more which
answer nothing hold the connection of the listener for the connection
timeout each. The watchdog acts at its deadline with the connection
still under way; an entry the stop publishes, which waits for the lock
of that connection, is not written to the session the connection opens,
and the stop ends once the connection does. | red | red |
| `aDropByTheReplicationServerAfterAGiveUpIsStillReported` | After a
change given up for credit, a replication server which drops the session
with no `StopMsg` is reported, and the stop ends. | red | n/a |

The fake replication server drops a session with `Session.abort()`,
which round 1 does not have: `aDropBy…` cannot run there. On master its
stop waits for credit and does not end. Round 2 (`26f7da4439`) fails
`aDropBy…` and `aConnectionUnderWay…`, and round 3 (`4d1130e44e`) fails
`aConnectionUnderWay…`, and only it.

`SessionTest` gains three cases, red on master and on round 1, which
have no `Session.abort()`; round 2 fails the third:

| Case | What it pins |
|---|---|
| `anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt` | A
write on the publishing thread, which a peer that reads nothing holds,
and a close which waits for the lock of the writes behind it, both end
with an abort; the peer finds the message cut short and the end of the
stream, with no `StopMsg`. |
| `anAbortEndsAWriteOfTheThreadOfTheSession` | A write of the thread of
the session, which any close joins, ends with an abort, and the thread
with it. |
| `anAbortMarksTheCloseBegunBeforeItClosesTheSockets` | The socket of
the session records, as it is closed, whether the close was marked begun
by then: an abort marks it before it closes the sockets, so that the
listener whose read their close ends finds it begun. |

Mutants, each run against the classes of round 4 (`d3a6c33741`):

| Mutant | Caught by |
|---|---|
| the stop not cleared by `start()` |
`aBrokerStartedAgainWaitsForCredit`, `aDisableEndsAWrite…`,
`aSessionRestart…`, `nothingIsWritten…` |
| the stop recorded only after the replica offline message |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`aDisableEnds…WaitsForCredit`, `aDropBy…`, `aSessionRestart…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…`,
`nothingIsWritten…` |
| no deadline: give up at the first wait without credit |
`aDisableEnds…NoCredit`, `aStopPublishes…`, `withNoConnectionTimeout…` |
| give-up not kept for the later messages | `aStopWhichOverlaps…`,
`aStopWritesNothing…`, `nothingIsWritten…` |
| a stopping broker writes nothing at all |
`aBrokerStartedAgainWaitsForCredit`, `aConnectionUnderWay…`,
`aDisableEnds…NoCredit`, `aStopEndsItsOwnWrite…`, `aStopPublishes…`,
`withNoConnectionTimeout…` |
| only the replica offline message given up | `aDropBy…`,
`aStopWhichOverlaps…`, `aStopWritesNothing…` |
| no `finally` around the replica offline message |
`aStopWhichCannotPublish…` |
| a connection timeout of 0 gives up at the first wait (round 0) |
`withNoConnectionTimeout…` |
| a connection timeout of 0 read as the shipped 5 seconds |
`withNoConnectionTimeout…` |
| a constant deadline of 5 seconds instead of the connection timeout |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`withNoConnectionTimeout…` |
| a constant deadline of 30 seconds | `aConnectionUnderWay…`,
`aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`aDisableEnds…WaitsForCredit`, `aDropBy…`, `aSessionRestart…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…`,
`nothingIsWritten…` |
| `ReplicationDomain.publish()` puts in the state only what the broker
wrote | `aSessionRestart…` |
| an overlapping stop replaces the one in progress |
`aStopWhichOverlaps…` |
| the give-up of a publish reported as a message written |
`aDisableEnds…NoCredit` (the announcement is left) |
| after a give-up, only the updates refused | `aConnectionUnderWay…`,
`aStopEndsItsOwnWrite…`, `aStopWritesNothing…` |
| no watchdog (round 1) | `aConnectionUnderWay…`, `aDisableEndsAWrite…`,
`aStopEndsItsOwnWrite…` |
| the watchdog closes the session but gives nothing up |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aStopEndsItsOwnWrite…`
(a session opened again) |
| the watchdog acts at the credit deadline | `aDisableEndsAWrite…`,
`aDisableEnds…NoCredit` (no `StopMsg`), `aDropBy…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…` |
| the watchdog acts whether the stop ended or not |
`aBrokerStartedAgainWaitsForCredit` |
| a watchdog with a connection timeout of 0 | `withNoConnectionTimeout…`
|
| the write deadline a constant 5 seconds past the credit deadline |
`aConnectionUnderWay…` (the watchdog had not acted),
`aDisableEndsAWrite…` |
| the write deadline three times the connection timeout from the start
of the stop | `aConnectionUnderWay…` (the watchdog had not acted),
`aDisableEndsAWrite…` |
| the write deadline two and a half times it | `aConnectionUnderWay…`
(the watchdog had not acted), `aDisableEndsAWrite…` |
| a session opened again once the stop gave up | `aDisableEndsAWrite…`,
`aDropBy…`, `aStopEndsItsOwnWrite…` |
| no read of the give-up under `connectPhaseLock` before the write, as
round 3 did | `aConnectionUnderWay…` |
| the listener reports no drop by the replication server once the stop
gave up, as round 2 did | `aDropBy…` |
| `Session.abort()` a plain close | `anAbortEndsAWriteThePeer…`,
`anAbortEndsAWriteOfTheThread…`, `aDisableEndsAWrite…`, `aDropBy…`,
`aStopEndsItsOwnWrite…`, `theHeartbeatMonitorEnds…` |
| `Session.abort()` closes the session before the sockets |
`anAbortEndsAWriteOfTheThreadOfTheSession` |
| `Session.abort()` leaves the sockets to the close | both `SessionTest`
cases of the abort |
| `Session.abort()` leaves a close which had begun before to wait |
`anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt` |
| the heartbeat monitor closes a session whose heartbeats stopped as one
with no error, as round 2 did | `theHeartbeatMonitorEnds…` |
| `Session.abort()` closes the sockets before it marks the close begun,
as round 2 did | `anAbortMarksTheCloseBegun…` |
| no second read of the stop before its give-up is marked | not caught -
see below |
| a window probe written after the give-up | not caught - see below |

Under the mutant which gives up only the replica offline message, a
local change which waits for credit under the lock never gives up, and
the watchdog ends it at twice the connection timeout: that is why that
mutant no longer fails the cases where such a change holds the stop, as
it did in round 1.

The mutant "the state only what the broker wrote" survives a disable
followed by `enable()`: `enable()` loads the state again, and
`PersistentServerState.checkAndUpdateServerState()` repairs the CSNs of
this replica from `ds-sync-hist`. A restart of the session keeps the
state in memory, and only there does the mutant lose the change.

Two races are not pinned, as a test cannot put a thread between the two
steps without a hook: a give-up marked after `start()` cleared the stop
it was read from - the mark lives in the object of that stop, and the
second read of the stop narrows the window to that of master's loop head
- and a watchdog acting after the end of its stop, which it does under
the monitor `StopInProgress.end()` takes, under `startStopLock`.

Nor is the start of the watchdog inside the `try`: a thread which fails
to start needs a hook as well. The read of the give-up before a window
probe is not pinned either: the probe is a few bytes which neither hold
the stop nor tell the replication server anything of the changes, and
the read only keeps `publish()` from writing anything after a give-up.

## Verification

- `ReplicationBrokerStopTest` and `SessionTest` on their own at round 4
(`d3a6c33741`, on `616d5f6c17`), on macOS: 15/15 four times and 6/6
three times. Master fails 12 cases of the first - all but
`aBrokerStartedAgain…`, `aStopPublishes…` and `withNoConnectionTimeout…`
- and the 3 cases of the abort of the second. Round 1 (`8fb5973e30`)
fails the 4 earlier cases the table marks red for it,
`aConnectionUnderWay…`, and `aDropBy…`, which cannot run there, and the
3 cases of the abort. Round 2 (`26f7da4439`) fails
`aConnectionUnderWay…`, `aDropBy…`, `theHeartbeatMonitorEnds…` and
`anAbortMarks…`. Round 3 (`4d1130e44e`) fails `aConnectionUnderWay…`
only. Every mutant above is caught, but the two noted; the runs a port
collision with a test run of another worktree voided were run again.
- On Linux in a container (JDK 21): `ReplicationBrokerStopTest` 15/15
twice, and `aConnectionUnderWay…`, `aDropBy…` and `aDisableEndsAWrite…`
ten times more.
- `ReplicationSynchronizationProviderConfiguration.xml` validates
against `admin.xsd`.
- Rebased onto `d954c4c53f` (#1204, #1211, #1219 and #1218 came in), at
`276317c33a`, whose commit on top of round 4 changes a comment only; on
macOS with JDK 21: `ReplicationBrokerStopTest` 15/15 and `SessionTest`
6/6, twice each, and the neighbours which share `publish()`:
`ReplicationBrokerTest` (#1218) 37/37 and `HistoricalCsnOrderingTest`
(#1204) 54/54.

 opendj-maven-plugin/src/main/resources/config/xml/org/forgerock/opendj/server/config/ReplicationSynchronizationProviderConfiguration.xml

@@ -14,6 +14,7 @@

  Copyright 2007-2008 Sun Microsystems, Inc.
  Portions Copyright 2011-2015 ForgeRock AS.
  Portions Copyright 2026 3A Systems, LLC.
  ! -->
<adm:managed-object name="replication-synchronization-provider"
  plural-name="replication-synchronization-providers"
@@ -97,8 +98,22 @@
  <adm:property name="connection-timeout" advanced="true">
    <adm:synopsis>
      Specifies the timeout used when connecting to peers and when performing
      SSL negotiation.
      SSL negotiation, and how long a stopping replication domain waits for
      window credit before it gives up the changes it still has to send.
    </adm:synopsis>
    <adm:description>
      A replication domain is stopped when it is disabled, when the server
      shuts down, and before an import, a restore or a restart of its session.
      A stopping domain gives its replication server this long to give the
      window credit for the changes it still has to send, and for the message
      which announces it offline: raising the value lengthens a stop against a
      replication server which gives no credit, lowering it shortens the time
      before those changes are given up. A replication server which does not
      read what the domain has already written to it gets as long again, after
      which the domain closes the session without waiting any longer. A value
      of 0 sets no limit: a connection, an SSL negotiation and a stopping
      replication domain then wait for as long as it takes.
    </adm:description>
    <adm:default-behavior>
      <adm:defined>
        <adm:value>5 seconds</adm:value>

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -2574,8 +2574,10 @@
       */
      logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
          + " could not announce itself offline: the message was not published - a change which"
          + " is still in flight holds it back, or the broker had no session to write it to,"
          + " and " + pendingChanges.size() + " change(s) are pending");
          + " is still in flight holds it back, the broker had no session to write it to, or it"
          + " got no window credit for it, or for a change before it, or the replication server"
          + " did not read what was written to it, in the time the stop gives the replication server"
          + " - and " + pendingChanges.size() + " change(s) are pending");
    }
  }


 opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java

@@ -194,18 +194,35 @@
  @Override
  public void close()
  {
    close(false);
  }

  /**
   * Closes the session, as {@link #close()} or as {@link #abort(IOException)} does.
   *
   * @param abort
   *          whether the sockets are to be closed at once, before anything waits for the
   *          publisher thread or for {@code publishLock} - even when a close had begun already
   */
  private void close(final boolean abort)
  {
    Throwable localSessionError;
    final boolean closing;

    synchronized (stateLock)
    {
      if (closeInitiated)
      {
        return;
      }

      closing = closeInitiated;
      localSessionError = sessionError;
      closeInitiated = true;
    }
    if (abort)
    {
      StaticUtils.close(plainSocket, secureSocket);
    }
    if (closing)
    {
      return;
    }

    try {
      interrupt();
@@ -319,6 +336,30 @@
    StaticUtils.close(plainSocket, secureSocket);
  }

  /**
   * Closes this session at once, as a session which failed with the error provided: nothing more
   * is written to it - neither what its publisher thread left queued nor the StopMsg - and a write
   * blocked in its socket, because the peer reads nothing, fails rather than holds the close.
   * <p>
   * {@link #close()} of a session with no error takes {@code publishLock} before it closes the
   * sockets, and a write blocked in the socket holds that lock for as long as the peer does not
   * read, which TCP does not bound; and any close first joins the publisher thread, which may be
   * the writer. So the error is recorded first, and the close is marked begun, so that a reader
   * whose read the close of the sockets ends finds it begun - see {@link #closeInitiated()}. Then
   * the sockets are closed, before the join: that ends the write, whichever thread makes it, and
   * releases a close which had begun before, with no error, and waits for the lock - the call here
   * then returns at once. The close itself takes the branch of a session which failed, and writes
   * nothing.
   *
   * @param cause
   *          why the session is given up, which it then reports as its error
   */
  public void abort(final IOException cause)
  {
    setSessionError(cause);
    close(true);
  }



  /**

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

@@ -146,8 +146,10 @@
   * period; the peers it is queued for, if they are recorded in that window, are lost too, with
   * the opposite effect - the first forward ends its wait, as for a message no peer was recorded
   * for. That window is the one publish the broker refuses: at once on a connection error or a
   * pending recovery, the broker's retry loop up to the reconnect when it has no session. The
   * wait it can cost is bounded by a grace period which is already running.
   * pending recovery, the broker's retry loop up to the reconnect when it has no session, and the
   * wait of a stopping broker for window credit, or for its replication server to read, up to the
   * deadline of the stop. The wait it can cost is bounded by a grace period which is already
   * running.
   *
   * @param baseDN
   *          the domain for which the message was announced

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

@@ -13,11 +13,15 @@
 *
 * Copyright 2007-2009 Sun Microsystems, Inc.
 * Portions Copyright 2011-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */

package org.opends.server.replication.service;

import static org.opends.messages.ReplicationMessages.*;

import java.io.IOException;

import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.protocol.Session;

@@ -107,6 +111,7 @@
                heartbeatInterval);
    }

    boolean heartbeatsStopped = false;
    try
    {
      boolean gotOneFailure = false;
@@ -124,6 +129,7 @@

            // Exit monitor and close session.
            shutdown = true;
            heartbeatsStopped = true;
            break;
          }
          else
@@ -161,7 +167,23 @@
      {
        logger.trace("Heartbeat monitor is exiting");
      }
      if (heartbeatsStopped)
      {
        /*
         * A replication server which is gone without a reset acknowledges nothing, and the writes to
         * it fill the socket buffers: a write then blocks in the socket, holding the lock a close of
         * a session with no error waits for, until TCP gives up retransmitting. Nothing else ends
         * that write, nor the read of the listener, which neither data nor a reset ends: the
         * session is closed as a failed one, with no StopMsg the replication server would not read.
         */
        session.abort(new IOException("No heartbeat came from replication server RS("
            + replicationServerID + ") at " + session.getReadableRemoteAddress() + " for twice "
            + heartbeatInterval + " ms"));
      }
      else
      {
      session.close();
    }
  }
}
}

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

@@ -51,6 +51,7 @@
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.server.ReplicationDomainCfg;
import org.forgerock.util.Utils;
import org.opends.server.api.DirectoryThread;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.DSInfo;
@@ -166,8 +167,91 @@
    }
  }

  /**
   * A stop of the broker, from the moment {@link #stop()} begins until {@link #start()} clears it.
   * <p>
   * What a publish gives up for the stop is marked on the stop, and not on the broker: a publish
   * which found the stop past its deadline may mark it only after start() cleared it, and then the
   * mark lands on a stop which is over, rather than on the broker started again.
   */
  private static final class StopInProgress
  {
    /**
     * The {@link System#nanoTime()} until which a publish waits for credit. The replication server
     * gets as long to give the credit as it gets to accept a connection: one which is only slow
     * takes the changes and the replica offline message as before, and only one which takes
     * nothing for that long costs them.
     */
    private final long creditDeadline;
    /**
     * The {@link System#nanoTime()} past which a stop still held is ended by the close of the
     * session - see {@link ReplicationBroker#endWhatStillHoldsTheStop(StopInProgress)}. It comes as
     * long after {@link #creditDeadline} as that comes after the start of the stop: a replication
     * server which reads slowly gets as long to read what was written as it got to give credit.
     */
    private final long writeDeadline;
    /**
     * Whether the stop has a deadline: a connection timeout of 0 lets a connection wait for as
     * long as it takes, and the stop as well, as it did before it had any deadline.
     */
    private final boolean bounded;
    /**
     * Set when a publish gave its message up because of this stop, or when the stop closed its
     * session: from then on publish() writes nothing, and the broker opens no session again, until
     * start() clears the stop. The change time heartbeats and the window and status messages
     * written straight to the session are not held back. A message published after one given up
     * would reach the replication server ahead of an older one - the replica offline message,
     * which carries the newest CSN of the replica, would tell it that nothing older is to come.
     */
    private volatile boolean gaveUp;
    /**
     * Set once stop() is over, by {@link #end()}. Guarded by this object, which the stop holds
     * while it closes the session, so that the close never comes after the end of the stop.
     */
    private boolean ended;
    /** The connection timeout the deadlines were set from, in milliseconds. */
    private final long connectionTimeoutMS;

    private StopInProgress(long connectionTimeoutMS)
    {
      final long timeout = TimeUnit.MILLISECONDS.toNanos(connectionTimeoutMS);
      this.creditDeadline = System.nanoTime() + timeout;
      this.writeDeadline = creditDeadline + timeout;
      this.bounded = connectionTimeoutMS > 0;
      this.connectionTimeoutMS = connectionTimeoutMS;
    }

    private boolean isPastTheCreditDeadline()
    {
      return bounded && System.nanoTime() - creditDeadline >= 0;
    }

    /** Ends the stop, once a close of its session in progress is done; none comes after it. */
    private synchronized void end()
    {
      ended = true;
      notifyAll();
    }
  }

  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
  private volatile boolean shutdown;
  /**
   * The stop in progress: set by {@link #stop()} before anything else it does, and cleared to
   * {@code null} by {@link #start()}. While it is set, a publish which finds no window credit by
   * its deadline gives its message up instead of waiting on.
   * <p>
   * stop() publishes the replica offline message before it sets {@link #shutdown}, the flag which
   * ends the wait for credit, and that message needs credit too, and the lock of the pending
   * changes a local change waits for credit under. Credit comes only as the replication server
   * takes the updates of the session, while its heartbeats keep the session up: without a bound a
   * replication server which takes no update holds the stop - and the disable, the restart or the
   * shutdown of the domain behind it - for as long as it does. A replication server which reads
   * nothing holds it as well, through a write which blocks in the socket once its buffers are full,
   * with the locks the write is made under: that one only the close of the socket ends, and the
   * stop closes the session past a second deadline.
   */
  private volatile StopInProgress stopping;
  private final Object startStopLock = new Object();
  private volatile ReplicationDomainCfg config;
  /** String reported under CSN=monitor when there is no connected RS. */
@@ -287,6 +371,7 @@
        return;
      }
      shutdown = false;
      stopping = null;
      this.rcvWindow = getMaxRcvWindow();
      connectAsDataServer();
    }
@@ -2220,6 +2305,17 @@
          break;
        }

        /*
         * Once the stop gave a message up, nothing more is published, and a session opened while the
         * stop runs would carry only its change time heartbeats - which tell the replication server
         * that nothing older is to come from this replica, while it still has the changes given up -
         * and hold the stop for as long as the connection takes: the stop waits for this lock. The
         * loop waits for the end of the stop instead - the listener gets here when the stop closes
         * the session it reads. A connection already under way when the stop gives up still opens
         * its session, which publish() then writes nothing to, and which the stop closes as it ends.
         */
        if (!gaveUpForTheStop())
        {
        try
        {
          connectAsDataServer();
@@ -2230,6 +2326,7 @@
          logger.error(NOTE_EXCEPTION_RESTARTING_SESSION,
              getBaseDN(), e.getLocalizedMessage() + " " + stackTraceToSingleLineString(e));
        }
        }

        if (rs.isConnected() || !infiniteTry)
        {
@@ -2319,7 +2416,7 @@
  {
    boolean done = false;

    while (!done && !shutdown)
    while (!done && !shutdown && !gaveUpForTheStop())
    {
      if (connectionError)
      {
@@ -2400,6 +2497,7 @@
          credit = true;
        }

        final StopInProgress stop = stopping;
        if (credit)
        {
          synchronized (connectPhaseLock)
@@ -2410,6 +2508,18 @@
            Check the session. If it has changed, some disconnection or
            reconnection happened and we need to restart from scratch.
            */
            if (gaveUpForTheStop())
            {
              /*
               * The stop gave up since the loop head, while this publish waited for this lock - for
               * a connection, say, which held it past the deadline of the watchdog of the stop, and
               * left it no session to close: the loop head decides, rather than this publish writes
               * to the session that connection opened. The watchdog marks the give-up before it
               * reads the session, and a connection sets its session before it releases this lock:
               * either the watchdog closes the session read here, or this read finds the mark.
               */
              continue;
            }
            final Session session = connectedRS.get().session;
            if (session != null && session == currentSession)
            {
@@ -2418,6 +2528,25 @@
            }
          }
        }
        else if (stop != null && stop.isPastTheCreditDeadline())
        {
          if (stop != stopping)
          {
            // start() ran since the read: the loop head decides, as before a stop had a deadline.
            continue;
          }
          /*
           * Only credit ends this wait on a session the replication server keeps up, and stop()
           * sets the flag of this loop only after the replica offline message, which may need the
           * lock this publish holds. What is given up here is a change which the next session
           * republishes - from the historical information of its entry, or a local delete, one the
           * recovery publishes included, from the deletes this domain keeps aside in memory, which
           * a restart of the server loses - or the replica offline message, which is then reported
           * as not sent.
           */
          stop.gaveUp = true;
          return false;
        }
        if (!credit && currentWindowSemaphore.availablePermits() == 0)
        {
          synchronized (connectPhaseLock)
@@ -2427,9 +2556,10 @@
            Send a WindowProbeMsg message to wake up the receiver in case the
            window update message was lost somehow...
            then loop to check again if connection was closed.
            Not once the stop gave up, which it may have done while this waited for the lock.
            */
            Session session = connectedRS.get().session;
            if (session != null)
            if (session != null && !gaveUpForTheStop())
            {
              session.publish(new WindowProbeMsg());
            }
@@ -2449,7 +2579,8 @@
        }

        // The receive threads should handle reconnection or
        // mark this broker in error. Just retry.
        // mark this broker in error. Just retry - unless the stop closed the session to end this
        // write: it marked the give-up first, and the loop ends on it.
        synchronized (connectPhaseLock)
        {
          try
@@ -2477,13 +2608,20 @@
      }
    }
    /*
     * The loop also ends when the broker is stopped, and then nothing was written to any
     * session: a caller told the message was published would report as sent a message which
     * never reached the wire.
     * The loop also ends when the broker is stopped, or gave a message up for the stop, and then
     * nothing was written to any session: a caller told the message was published would report as
     * sent a message which never reached the wire.
     */
    return done;
  }

  /** Tells whether a publish gave its message up for the stop in progress, if there is one. */
  private boolean gaveUpForTheStop()
  {
    final StopInProgress stop = stopping;
    return stop != null && stop.gaveUp;
  }

  /**
   * Receive a message.
   * This method is not thread-safe and should either always be
@@ -2765,22 +2903,115 @@
    {
      debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
    }
    final StopInProgress stop;
    final boolean newStop;
    synchronized (startStopLock) {
      if (shutdown) {
        return;
      }
      // A stop which overlaps one in progress keeps its deadlines and what it gave up.
      newStop = stopping == null;
      if (newStop)
      {
        stopping = new StopInProgress(MultimasterReplication.getConnectionTimeoutMS());
      }
      stop = stopping;
    }

    try
    {
      // In the try: a watchdog which cannot be started must not leave the broker stopping either.
      if (newStop && stop.bounded)
      {
        new DirectoryThread(() -> endWhatStillHoldsTheStop(stop),
            "Replica DS(" + getServerId() + ") watchdog of the stop for domain \"" + getBaseDN() + "\"")
            .start();
      }
    domain.publishReplicaOfflineMsg();

    }
    finally
    {
      // A broker left running as a stopping one would give up every publish which waits for credit.
    synchronized (startStopLock)
    {
        try
        {
      shutdown = true;
      setConnectedRS(ConnectedRS.stopped());
      stopRSHeartBeatMonitoring();
      stopChangeTimeHeartBeatPublishing();
      deregisterReplicationMonitor();
    }
        finally
        {
          // Under the lock, so that no start() opens a session which this stop could still close.
          stop.end();
        }
      }
    }
  }

  /**
   * Waits for the stop provided to end, and closes the session if the stop is still held once it is
   * past its {@link StopInProgress#writeDeadline}.
   * <p>
   * A publish which waits for credit gives up by itself once the stop is past its credit deadline,
   * and the stop then ends the session in order. What holds the stop long after that is a write the
   * replication server does not read - of a change which got its credit, of a window probe, of the
   * StopMsg which ends the session - or a lock which such a write holds: the lock of the pending
   * changes, which the replica offline message needs, the lock of the connection phase, or the lock
   * of the writes of the session, which a close of a session with no error takes. A socket write
   * has no timeout, and returns only once the peer reads or the socket is closed: the session is
   * closed here as a failed one, without that lock - see {@link Session#abort(IOException)}. The
   * give-up is marked first, so that a publish whose write then fails returns rather than retries,
   * and no session is opened again for the stop.
   * <p>
   * The close is made under the monitor of the stop, which {@link StopInProgress#end()} takes, and
   * stop() ends before start() can run again: the session closed here is never one of a broker
   * started again.
   */
  private void endWhatStillHoldsTheStop(final StopInProgress stop)
  {
    synchronized (stop)
    {
      long left;
      while (!stop.ended && (left = stop.writeDeadline - System.nanoTime()) > 0)
      {
        try
        {
          TimeUnit.NANOSECONDS.timedWait(stop, left);
        }
        catch (InterruptedException e)
        {
          // Only a shutdown of the server interrupts this thread: the stop is left to itself.
          Thread.currentThread().interrupt();
          return;
        }
      }
      if (stop.ended)
      {
        return;
      }

      stop.gaveUp = true;
      /*
       * Read after the mark: a connection under way leaves no session to close here, and publish()
       * writes nothing to the one it then opens, as it reads the mark again under the lock that
       * connection holds. The stop closes that session as it ends.
       */
      final ConnectedRS rs = connectedRS.get();
      if (rs.session != null)
      {
        logger.warn(LocalizableMessage.raw("The stop of the replication domain %s of directory "
            + "server %d still waits for its replication server RS(%d) at %s after %d ms, twice "
            + "the connection timeout: its session is closed as a failed one, and nothing more is "
            + "written to it",
            getBaseDN(), getServerId(), rs.getServerId(), rs.replicationServer,
            2 * stop.connectionTimeoutMS));
        rs.session.abort(new IOException("The stop of the domain gave the replication server "
            + 2 * stop.connectionTimeoutMS + " ms to take what was written to the session"));
      }
    }
  }

  /**

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

@@ -3991,7 +3991,11 @@
   * @return {@code true} if the message was written to the session of the
   *         Replication Service, {@code false} if the broker could not write
   *         it: it has no usable session, the changes which come before this
   *         one still have to be republished, or it was stopped in between.
   *         one still have to be republished, or it was stopped in between -
   *         or is being stopped, and the window of its session gave it no
   *         credit for this message, or for one before it, or the replication
   *         server did not read what was written to it, in the time the stop
   *         gives the replication server.
   */
  public boolean publish(UpdateMsg msg)
  {

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplicationBrokerStopTest.java

New file
@@ -0,0 +1,1727 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static java.util.concurrent.TimeUnit.*;

import static org.forgerock.opendj.ldap.ModificationType.*;
import static org.opends.messages.ReplicationMessages.WARN_REPLICATION_SERVER_BADLY_DISCONNECTED;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.protocols.internal.Requests.*;
import static org.testng.Assert.*;

import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.TestCaseUtils;
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.protocol.DeleteMsg;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
import org.opends.server.replication.protocol.HeartbeatMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.ServerStartMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StartSessionMsg;
import org.opends.server.replication.protocol.StopMsg;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.protocol.WindowMsg;
import org.opends.server.replication.protocol.WindowProbeMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.TestTimer;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

/**
 * Tests that the broker of a domain can be stopped while its replication server keeps the session
 * up and takes no update from it, and that the stop gives up no more than it has to.
 * <p>
 * The replication server returns window credit as its reader takes the updates of the session, and
 * the same reader answers a window probe, while its heartbeats come from a thread of their own: a
 * reader held up leaves the window closed on a session the domain sees alive. A publish then waits
 * for credit with nothing but the stop of the broker to end it, and stop() publishes the replica
 * offline message - which needs credit as well, and the lock of the pending changes a local change
 * is pushed under - before it sets the flag which ends that wait.
 * <p>
 * A replication server which reads nothing of the session holds the stop as well, once the socket
 * buffers are full: a write then blocks in the socket, with the locks it is made under, and only
 * the close of the socket ends it. A replication server which is gone altogether, heartbeats
 * included, holds that write with no stop at all: the heartbeat monitor has to close the session
 * the same way.
 * <p>
 * The stop gives the replication server the connection timeout of the replication provider to give
 * credit, and as long again to read what was written. The suite sets it to 30 seconds: the cases run
 * with {@link #SHORT_CONNECTION_TIMEOUT_MS}, and those which need another one set it for their own
 * run.
 */
@SuppressWarnings("javadoc")
public class ReplicationBrokerStopTest extends ReplicationTestCase
{
  private static final int DS_ID = 1;
  private static final int OTHER_DS_ID = 3;
  private static final DN PROVIDER_DN =
      DN.valueOf("cn=Multimaster Synchronization,cn=Synchronization Providers,cn=config");
  /** The connection timeout the cases run with, unless they set their own. */
  private static final long SHORT_CONNECTION_TIMEOUT_MS = 2000;
  /**
   * A connection timeout longer than the shipped default of 5 seconds, and far longer than what a
   * replication server which is only slow needs in these cases to give credit.
   */
  private static final long LONG_CONNECTION_TIMEOUT_MS = 6000;
  /** The window of a replication server whose cases need no wait for credit: the default one. */
  private static final int LARGE_WINDOW = 100000;
  /**
   * The size of the messages a case writes until the socket buffers are full, in bytes or in
   * characters: a few dozen of them fill the buffers a loopback connection gets on its own.
   */
  private static final int LARGE_MESSAGE_SIZE = 256 * 1024;
  /** How many of those messages a case writes at most before it reports that no write blocks. */
  private static final int MAX_LARGE_MESSAGES = 400;
  /**
   * The heartbeat interval of the case of a replication server which is gone: long enough for its
   * writes to fill the socket buffers before the heartbeat monitor gives up on the session.
   */
  private static final long HEARTBEAT_INTERVAL_MS = 4000;
  /**
   * The replication servers which answer no connection in the case of a connection under way: each
   * costs the connection the connection timeout, and together they hold it well past twice that, the
   * deadline of the watchdog of a stop.
   */
  private static final int SILENT_SERVERS = 4;
  /** A value which makes a change of the description one of those messages. */
  private static final String LARGE_VALUE = "x".repeat(LARGE_MESSAGE_SIZE);

  private final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
  private long connectionTimeoutOfTheSuiteMS;

  @BeforeClass
  public void shortenTheConnectionTimeout()
  {
    connectionTimeoutOfTheSuiteMS = MultimasterReplication.getConnectionTimeoutMS();
    setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
  }

  @AfterClass(alwaysRun = true)
  public void restoreTheConnectionTimeout()
  {
    setConnectionTimeout(connectionTimeoutOfTheSuiteMS);
  }

  /**
   * With the only credit of the window spent, the replica offline message stop() publishes finds
   * none: the disable has to end all the same, and the message is given up rather than written -
   * once the replication server had the connection timeout, as it is configured, to give credit.
   * The announcement of the message is withdrawn, and the session is ended in order: a wait for
   * credit, which gives up by itself, is no reason to close it as a failed one.
   */
  @Test(timeOut = 120_000)
  public void aDisableEndsWhenTheReplicationServerGivesNoCredit() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);

      // The only credit of the window goes to a change of our own.
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final long beforeTheDisable = System.nanoTime();
      disabling = startThread("disable", domain::disable);
      disabling.join(stopBoundMS());
      final long disableMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable);
      assertFalse(disabling.isAlive(),
          "disable() waits for window credit a replication server which takes no update does not give");
      assertTrue(disableMS >= LONG_CONNECTION_TIMEOUT_MS, "disable() gave the replication server " + disableMS
          + " ms to give credit, not the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(ReplicaOfflineMsg.class), 0, "the replica offline message was written with no credit");
      assertEquals(rs.count(StopMsg.class), 1, "the session was closed as a failed one, not ended in order");

      /* The grace period of the announcement has run out by now, so whether the shutdown of a
       * collocated replication server would wait for it says nothing: what is left announced does.
       * An announcement older than the change is one of an earlier stop of the replica. */
      final CSN announced = offlineAnnouncementOfTheReplica();
      assertTrue(announced == null || !announced.isNewerThan(rs.receivedUpdateCSNs().get(0)),
          "the replica offline message given up is still announced: " + announced);
    }
    finally
    {
      cleanUp(rs, disabling);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * With no connection timeout, a connection waits for its replication server for as long as it
   * takes, and so does a stop for credit: the replica offline message waits past its first wait for
   * credit and past the deadline a stop would have with the shipped default of 5 seconds, and it
   * goes out once the credit comes.
   */
  @Test(timeOut = 120_000)
  public void withNoConnectionTimeoutAStopWaitsForTheCredit() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    setConnectionTimeout(0);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final long beforeTheDisable = System.nanoTime();
      disabling = startThread("disable", domain::disable);
      // A probe follows each wait for credit which found none, unless the publish gave up after it.
      rs.waitForWindowProbes(2);
      Thread.sleep(Math.max(0,
          LONG_CONNECTION_TIMEOUT_MS + 1000 - NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable)));
      assertTrue(disabling.isAlive(), "with no connection timeout, the stop gave the replica offline message up");

      rs.giveCredit(1);
      disabling.join(stopBoundMS());
      assertFalse(disabling.isAlive(), "disable() did not end once the replication server gave credit");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
    }
    finally
    {
      cleanUp(rs, disabling);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * A local change which waits for credit holds the lock of the pending changes, which stop() needs
   * to publish the replica offline message: the disable has to end the wait rather than wait for
   * it, and the change ends with it.
   */
  @Test(timeOut = 120_000)
  public void aDisableEndsWhileALocalChangeWaitsForCredit() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread writer = null;
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final AtomicReference<ResultCode> second = new AtomicReference<>();
      writer = startThread("writer", () -> second.set(replaceDescription("second")));
      // A probe is sent only by a publish which waited for credit and found the window closed.
      rs.waitForWindowProbes(1);

      disabling = startThread("disable", domain::disable);
      disabling.join(stopBoundMS());
      assertFalse(disabling.isAlive(),
          "disable() waits for a local change which waits for window credit under the lock it needs");
      writer.join(stopBoundMS());
      assertFalse(writer.isAlive(), "the local change still waits for credit on a stopped broker");
      assertEquals(second.get(), ResultCode.SUCCESS);
    }
    finally
    {
      cleanUp(rs, disabling, writer);
    }
  }

  /**
   * A replication server which is only slow gives the credit before the stop gives up on it: the
   * change which waited for credit under the lock goes out, and the replica offline message after
   * it, as they did before a stop could give anything up.
   */
  @Test(timeOut = 120_000)
  public void aStopPublishesWhatTheReplicationServerTakesInTime() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    // The credit comes about a second into the stop: well within the time the stop gives.
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread writer = null;
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final AtomicReference<ResultCode> second = new AtomicReference<>();
      writer = startThread("writer", () -> second.set(replaceDescription("second")));
      rs.waitForWindowProbes(1);
      disabling = startThread("disable", domain::disable);
      waitForTheOfflineMessageToWaitForTheLock(disabling);

      // The change keeps waiting for credit well into the stop: two more waits of it.
      rs.waitForWindowProbes(rs.count(WindowProbeMsg.class) + 2);
      rs.giveCredit(2);

      disabling.join(stopBoundMS());
      assertFalse(disabling.isAlive(), "disable() did not end once the replication server gave credit");
      writer.join(stopBoundMS());
      assertEquals(second.get(), ResultCode.SUCCESS);
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(LDAPUpdateMsg.class), 2, "the change which waited for credit was given up");
      assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
      final List<Class<?>> updates = rs.receivedUpdates();
      assertEquals(updates.get(updates.size() - 1), ReplicaOfflineMsg.class,
          "the replica offline message has to come after the changes before it: " + updates);
    }
    finally
    {
      cleanUp(rs, disabling, writer);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * Once the stop gave a change up, the broker writes nothing more: the replica offline message
   * carries the newest CSN of the replica, and written after a change it gave up it tells the
   * replication server that the replica has nothing more to send up to that CSN. A broker started
   * again finds the replica ahead of its replication server, and the recovery of its session
   * publishes the change given up; then it writes as before.
   */
  @Test(timeOut = 120_000)
  public void nothingIsWrittenAfterAChangeTheStopGaveUp() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    // The stop waits for the lock until the change gives up: long enough for the case to see it.
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread writer = null;
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final AtomicReference<ResultCode> second = new AtomicReference<>();
      writer = startThread("writer", () -> second.set(replaceDescription("second")));
      rs.waitForWindowProbes(1);
      disabling = startThread("disable", domain::disable);
      waitForTheOfflineMessageToWaitForTheLock(disabling);

      /* The change is given up, and credit comes right after it: a stop which went on writing would
       * still wait for it, while one which writes nothing may have ended the session already. */
      writer.join(stopBoundMS());
      assertFalse(writer.isAlive(), "the local change still waits for credit on a stopped broker");
      rs.giveCreditIfTheSessionIsUp(1);

      disabling.join(stopBoundMS());
      assertFalse(disabling.isAlive(), "disable() waits for window credit");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(LDAPUpdateMsg.class), 1);
      assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
          "the replica offline message was written after a change the stop gave up");

      /* That is over once the broker is started again. The replication server has only the first
       * change, and the domain the second as well: the recovery of the new session republishes it,
       * with the credit of the window of that session. */
      domain.enable();
      waitForConnection(domain);
      rs.waitForUpdates(2);
      final List<CSN> csns = rs.receivedUpdateCSNs();
      assertTrue(csns.get(1).isNewerThan(csns.get(0)) && csns.get(1).getServerId() == DS_ID,
          "the change the stop gave up was not republished: " + csns);
      waitForTheRecoveryToEnd();

      rs.giveCredit(1);
      assertEquals(replaceDescription("third"), ResultCode.SUCCESS);
      rs.waitForUpdates(3);
    }
    finally
    {
      cleanUp(rs, disabling, writer);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * A restart of the session stops the broker as a disable does, and the change the stop gives up
   * is republished by the recovery of the next session. Unlike enable(), which loads the state of
   * the domain again and repairs it from the historical information of the entries, a restart
   * keeps the state in memory: the change is known to be missing only if that state took it
   * although the broker did not write it.
   */
  @Test(timeOut = 120_000)
  public void aSessionRestartRepublishesTheChangeTheStopGaveUp() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread writer = null;
    Thread restarting = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final AtomicReference<ResultCode> second = new AtomicReference<>();
      writer = startThread("writer", () -> second.set(replaceDescription("second")));
      rs.waitForWindowProbes(1);
      restarting = startThread("restart", domain::restartService);
      restarting.join(stopBoundMS());
      assertFalse(restarting.isAlive(), "the restart of the session waits for window credit");
      writer.join(stopBoundMS());
      assertEquals(second.get(), ResultCode.SUCCESS);

      waitForConnection(domain);
      rs.waitForUpdates(2);
      final List<CSN> csns = rs.receivedUpdateCSNs();
      assertTrue(csns.get(1).isNewerThan(csns.get(0)) && csns.get(1).getServerId() == DS_ID,
          "the change the stop gave up was not republished: " + csns);
      assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
          "the replica offline message was written after a change the stop gave up");
      waitForTheRecoveryToEnd();

      rs.giveCredit(1);
      assertEquals(replaceDescription("third"), ResultCode.SUCCESS);
      rs.waitForUpdates(3);
    }
    finally
    {
      cleanUp(rs, restarting, writer);
    }
  }

  /**
   * A stop gives up only what finds no credit: with the window open, the replica offline message
   * still goes out. And once started again, the broker waits for credit as it did before it was ever
   * stopped: a stop does not leave behind a broker which gives a publish up as soon as the window is
   * closed, nor one whose session is closed once the stop is past the deadline it had.
   */
  @Test(timeOut = 120_000)
  public void aBrokerStartedAgainWaitsForCredit() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    Thread writer = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      domain.disable();
      final long stopped = System.nanoTime();
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(ReplicaOfflineMsg.class), 1, "the replica offline message was given up");
      domain.enable();
      waitForConnection(domain);

      assertEquals(replaceDescription("first"), ResultCode.SUCCESS);
      rs.waitForUpdates(1);

      final AtomicReference<ResultCode> second = new AtomicReference<>();
      writer = startThread("writer", () -> second.set(replaceDescription("second")));
      // Two probes: the publish waited for credit twice, and did not give up after the first wait.
      rs.waitForWindowProbes(2);
      /* Nor past the time the stop gave the replication server to give credit and then to read: a
       * flag the stop left would end the wait by then, and a close the stop left due the session. */
      final long pastTheStopMS = 2 * MultimasterReplication.getConnectionTimeoutMS() + 1000
          - NANOSECONDS.toMillis(System.nanoTime() - stopped);
      Thread.sleep(Math.max(0, pastTheStopMS));
      assertTrue(writer.isAlive(), "a publish gave the change up while nobody stops the broker");
      assertEquals(rs.sessionsStarted(), 2, "the session of the broker started again was closed");

      rs.giveCredit(1);
      rs.waitForUpdates(2);
      writer.join(stopBoundMS());
      assertFalse(writer.isAlive(), "the local change still waits once the credit came");
      assertEquals(second.get(), ResultCode.SUCCESS);
    }
    finally
    {
      cleanUp(rs, writer);
    }
  }

  /**
   * A stop ends with the broker stopped even when publishing the replica offline message fails:
   * a broker left running as a stopping one would give up every publish which waits for credit.
   */
  @Test(timeOut = 120_000)
  public void aStopWhichCannotPublishTheOfflineMessageStillStopsTheBroker() throws Exception
  {
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    rs.start();
    final RuntimeException failure = new IllegalStateException("cannot publish the replica offline message");
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        throw failure;
      }
    }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
    try
    {
      connect(broker, 0);
      try
      {
        broker.stop();
        fail("the failure of the replica offline message was swallowed");
      }
      catch (IllegalStateException e)
      {
        assertSame(e, failure);
      }
      assertFalse(broker.isConnected(), "the broker still runs its session after it was stopped");
    }
    finally
    {
      rs.shutdown();
    }
  }

  /**
   * Once a publish gave its message up for the stop, the next one is refused even with the window
   * open: the stop publishes a change which finds no credit by its deadline, then waits for the
   * credit the replication server gives right after, and publishes a second change, which a broker
   * that only gave up the first would write. Nor do the messages of a total update, which need no
   * credit, go out: an import would be told it was done with an entry missing.
   */
  @Test(timeOut = 120_000)
  public void aStopWritesNothingAfterAGiveUpEvenWithCredit() throws Exception
  {
    // The stop waits for the case to read the credit: well before the session is closed for it.
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    rs.start();
    final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
    final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
    final AtomicBoolean publishedWithCredit = new AtomicBoolean();
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        final ReplicationBroker stopping = brokerRef.get();
        stopping.publish(newDeleteMsg(csns));
        try
        {
          rs.giveCredit(1);
          readTheCredit(stopping);
        }
        catch (Exception e)
        {
          throw new IllegalStateException(e);
        }
        stopping.publish(newDeleteMsg(csns));
        stopping.publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID, new byte[16], 1));
        stopping.publish(new DoneMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID));
        publishedWithCredit.set(true);
      }
    }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
    brokerRef.set(broker);
    Thread stopping = null;
    try
    {
      // A read of the session which finds nothing returns in a while rather than never.
      connect(broker, 200);
      // The only credit of the window goes to a first change.
      broker.publish(newDeleteMsg(csns));
      rs.waitForUpdates(1);

      stopping = startThread("stop", broker::stop);
      stopping.join(stopBoundMS());
      assertFalse(stopping.isAlive(), "stop() waits for window credit a replication server which takes no update does not give");
      assertTrue(publishedWithCredit.get(), "the stop did not get to publish a change with the window open");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(LDAPUpdateMsg.class), 1, "a change was written after one the stop gave up");
      assertEquals(rs.count(EntryMsg.class) + rs.count(DoneMsg.class), 0,
          "a message of a total update was written after a change the stop gave up");
    }
    finally
    {
      cleanUp(rs, stopping);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * A stop which overlaps one in progress keeps what that one gave up: once the first stop gave a
   * change up, it writes nothing more, even with the window open again, while a second stop runs
   * beside it - which a stop of its own, with no give-up yet, would let out.
   */
  @Test(timeOut = 120_000)
  public void aStopWhichOverlapsOneInProgressKeepsItsGiveUp() throws Exception
  {
    // The first stop waits for the case to read the credit: well before the session is closed for it.
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    rs.start();
    final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
    final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
    final AtomicInteger stops = new AtomicInteger();
    final CountDownLatch secondStopRuns = new CountDownLatch(1);
    final CountDownLatch firstStopIsDone = new CountDownLatch(1);
    final AtomicReference<Thread> secondStop = new AtomicReference<>();
    final AtomicBoolean publishedWithCredit = new AtomicBoolean();
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        if (stops.incrementAndGet() > 1)
        {
          // The second stop waits beside the first until that one is done.
          secondStopRuns.countDown();
          await(firstStopIsDone);
          return;
        }
        try
        {
          final ReplicationBroker stopping = brokerRef.get();
          stopping.publish(newDeleteMsg(csns));
          secondStop.set(startThread("second stop", stopping::stop));
          await(secondStopRuns);
          rs.giveCredit(1);
          readTheCredit(stopping);
          stopping.publish(newDeleteMsg(csns));
          publishedWithCredit.set(true);
        }
        catch (Exception e)
        {
          throw new IllegalStateException(e);
        }
        finally
        {
          firstStopIsDone.countDown();
        }
      }
    }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
    brokerRef.set(broker);
    Thread stopping = null;
    try
    {
      // A read of the session which finds nothing returns in a while rather than never.
      connect(broker, 200);
      // The only credit of the window goes to a first change.
      broker.publish(newDeleteMsg(csns));
      rs.waitForUpdates(1);

      stopping = startThread("first stop", broker::stop);
      stopping.join(stopBoundMS());
      assertFalse(stopping.isAlive(), "the first stop did not end");
      assertTrue(publishedWithCredit.get(), "the first stop did not get to publish a change with the window open");
      secondStop.get().join(stopBoundMS());
      assertFalse(secondStop.get().isAlive(), "the second stop did not end");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(LDAPUpdateMsg.class), 1,
          "a change was written after one the stop gave up, once a second stop had begun");
    }
    finally
    {
      cleanUp(rs, stopping, secondStop.get());
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * A replication server which keeps the session up but reads nothing of it holds the write of a
   * local change in the socket once the buffers are full, under the lock of the pending changes
   * stop() needs, and with credit to spare there is no wait for credit for the stop to give up: the
   * disable has to end the write by closing the session - once the replication server had the
   * connection timeout twice over - and the change comes back with the recovery of the next session.
   */
  @Test(timeOut = 120_000)
  public void aDisableEndsAWriteToAReplicationServerWhichReadsNothing() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    setConnectionTimeout(LONG_CONNECTION_TIMEOUT_MS);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
    final AtomicBoolean keepWriting = new AtomicBoolean(true);
    final AtomicInteger changes = new AtomicInteger();
    final AtomicReference<ResultCode> failed = new AtomicReference<>();
    Thread writer = null;
    Thread disabling = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();
      rs.stopReading();
      writer = startLargeChanges(keepWriting, changes, failed);
      waitForAWriteToBlock(writer, changes);
      keepWriting.set(false);

      final long beforeTheDisable = System.nanoTime();
      disabling = startThread("disable", domain::disable);
      disabling.join(stopBoundMS());
      final long disableMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheDisable);
      assertFalse(disabling.isAlive(), "disable() waits for a write the replication server does not read");
      assertTrue(disableMS >= 2 * LONG_CONNECTION_TIMEOUT_MS, "disable() gave the replication server "
          + disableMS + " ms, not twice the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
      // Half a connection timeout above: the bound of the join leaves room for three times it.
      assertTrue(disableMS < 2 * LONG_CONNECTION_TIMEOUT_MS + LONG_CONNECTION_TIMEOUT_MS / 2,
          "disable() gave the replication server " + disableMS
          + " ms, more than two and a half times the connection timeout of " + LONG_CONNECTION_TIMEOUT_MS + " ms");
      // The disable joined the listener, whose read the close of the session ended.
      assertEquals(badDisconnectionsLogged(), 0,
          "the close of the session by the stop was reported as one by the replication server");
      writer.join(stopBoundMS());
      assertFalse(writer.isAlive(), "the local change is still blocked in its write");
      assertNull(failed.get(), "a local change failed");

      // The change cut short ends what the replication server reads, and nothing comes after it.
      rs.resumeReadingTheSessionTheDomainClosed();
      rs.waitForTheSessionToEnd();
      final List<CSN> beforeTheEnable = rs.receivedUpdateCSNs();
      assertEquals(rs.count(ReplicaOfflineMsg.class), 0,
          "the replica offline message was written after a change the stop gave up");
      assertEquals(rs.sessionsStarted(), 1, "a session was opened again while the domain was being stopped");

      /* The disable left the state of the domain in its backend only, and the enable loads it: its
       * newest change of this replica is the one the stop gave up, as no other change came after it. */
      domain.enable();
      waitForConnection(domain);
      final CSN blocked = domain.getServerState().getCSN(DS_ID);
      assertNotNull(blocked, "the domain has no change of its own in its state");
      assertFalse(beforeTheEnable.contains(blocked), "the change blocked in its write was written: " + beforeTheEnable);
      newTimer().repeatUntilSuccess(() -> assertTrue(rs.receivedUpdateCSNs().contains(blocked),
          "the change the stop gave up, " + blocked + ", was not republished: the replication server has "
          + rs.receivedUpdateCSNs()));
    }
    finally
    {
      cleanUp(rs, disabling, writer);
      setConnectionTimeout(SHORT_CONNECTION_TIMEOUT_MS);
    }
  }

  /**
   * The write which blocks can be one of the stop itself, with no lock of the pending changes in
   * the way: the stop ends it the same way. The close of the session wakes the listener, which then
   * waits for the end of the stop rather than opens a session again: the broker writes nothing more
   * once the stop gave up, and the stop would wait for that connection.
   */
  @Test(timeOut = 120_000)
  public void aStopEndsItsOwnWriteAndOpensNoSessionAgain() throws Exception
  {
    final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
    rs.start();
    final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
    final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
    final AtomicReference<Thread> listenerRef = new AtomicReference<>();
    final AtomicInteger entries = new AtomicInteger();
    final AtomicInteger sessionsWhenTheListenerWaits = new AtomicInteger(-1);
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        final ReplicationBroker stopping = brokerRef.get();
        // Entries need no credit: they are written until one blocks, and the stop ends that one.
        long publishMS;
        do
        {
          final long beforeThePublish = System.nanoTime();
          stopping.publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID,
              new byte[LARGE_MESSAGE_SIZE], entries.incrementAndGet()));
          publishMS = NANOSECONDS.toMillis(System.nanoTime() - beforeThePublish);
        }
        while (publishMS < SHORT_CONNECTION_TIMEOUT_MS && entries.get() < MAX_LARGE_MESSAGES);
        try
        {
          newTimer().repeatUntilSuccess(() -> assertTrue(rs.sessionsStarted() > 1
              || waitsIn(listenerRef.get(), ReplicationBroker.class, "reStart"),
              "the listener neither opened a session again nor waits for the end of the stop"));
        }
        catch (Exception e)
        {
          throw new IllegalStateException(e);
        }
        sessionsWhenTheListenerWaits.set(rs.sessionsStarted());
        stopping.publish(newDeleteMsg(csns));
      }
    }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
    brokerRef.set(broker);
    Thread listener = null;
    Thread stopping = null;
    try
    {
      connect(broker, 0);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();
      listener = startThread("listener", () -> {
        try
        {
          while (broker.receive() != null)
          {
            // What the replication server sends needs no answer here.
          }
        }
        catch (SocketTimeoutException e)
        {
          throw new IllegalStateException(e);
        }
      });
      listenerRef.set(listener);
      rs.stopReading();

      final long beforeTheStop = System.nanoTime();
      stopping = startThread("stop", broker::stop);
      stopping.join(stopBoundMS());
      final long stopMS = NANOSECONDS.toMillis(System.nanoTime() - beforeTheStop);
      assertFalse(stopping.isAlive(), "stop() waits for a write the replication server does not read");
      assertTrue(entries.get() < MAX_LARGE_MESSAGES, "no write blocked after " + entries.get() + " entries");
      assertTrue(stopMS >= 2 * SHORT_CONNECTION_TIMEOUT_MS, "stop() gave the replication server " + stopMS
          + " ms, not twice the connection timeout of " + SHORT_CONNECTION_TIMEOUT_MS + " ms");
      assertEquals(sessionsWhenTheListenerWaits.get(), 1,
          "the listener opened a session again while the broker was being stopped");
      listener.join(stopBoundMS());
      assertFalse(listener.isAlive(), "the listener did not end with the stop");
      assertEquals(badDisconnectionsLogged(), 0,
          "the close of the session by the stop was reported as one by the replication server");

      rs.resumeReadingTheSessionTheDomainClosed();
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(LDAPUpdateMsg.class), 0, "a change was written after the entry the stop gave up");
      assertFalse(broker.isConnected(), "the broker still runs its session after it was stopped");
    }
    finally
    {
      cleanUp(rs, stopping, listener);
    }
  }

  /**
   * A connection under way when the watchdog of the stop gives up leaves it no session to close: the
   * listener lost the session in the stop, and connects again holding the lock of the connection
   * phase, while replication servers which take a connection but answer none cost it the connection
   * timeout each. A publish of the stop which waited for that lock must not write to the session the
   * connection then opens: one which the replication server does not read would hold the stop, with
   * nothing left to end it. The entry here goes to a replication server which reads, and must not be
   * written at all.
   */
  @Test(timeOut = 120_000)
  public void aConnectionUnderWayWhenTheStopGivesUpGetsNothingWritten() throws Exception
  {
    final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
    rs.start();
    final List<SilentReplicationServer> silentServers = new ArrayList<>();
    final SortedSet<String> urls = new TreeSet<>(Collections.singleton("127.0.0.1:" + rs.getPort()));
    for (int i = 0; i < SILENT_SERVERS; i++)
    {
      final SilentReplicationServer silent = new SilentReplicationServer();
      silent.start();
      silentServers.add(silent);
      urls.add("127.0.0.1:" + silent.getPort());
    }
    final DomainFakeCfg config = new DomainFakeCfg(baseDN, OTHER_DS_ID, urls);
    config.setHeartbeatInterval(100000);
    final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
    final AtomicReference<Thread> listenerRef = new AtomicReference<>();
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        try
        {
          for (SilentReplicationServer silent : silentServers)
          {
            silent.answerNothing();
          }
          // The replication server ends the session in order, and the listener connects again.
          rs.closeTheSession();
          newTimer().repeatUntilSuccess(() -> assertTrue(
              isIn(listenerRef.get(), ReplicationBroker.class, "collectReplicationServersInfo"),
              "the listener does not connect again"));
        }
        catch (Exception e)
        {
          throw new IllegalStateException(e);
        }
        // Waits for the lock the connection holds.
        brokerRef.get().publish(new EntryMsg(OTHER_DS_ID, WithholdingReplicationServer.RS_ID, new byte[16], 1));
      }
    }, new ServerState(), config, getReplSessionSecurity());
    brokerRef.set(broker);
    Thread listener = null;
    Thread stopping = null;
    try
    {
      connect(broker, 0);
      listener = startThread("listener", () -> {
        try
        {
          while (broker.receive() != null)
          {
            // What the replication server sends needs no answer here.
          }
        }
        catch (SocketTimeoutException e)
        {
          throw new IllegalStateException(e);
        }
      });
      listenerRef.set(listener);

      final long beforeTheStop = System.nanoTime();
      final Thread stop = startThread("stop", broker::stop);
      stopping = stop;
      newTimer().repeatUntilSuccess(() -> assertTrue(
          stop.getState() == Thread.State.BLOCKED && isIn(stop, ReplicationBroker.class, "publish"),
          "the entry of the stop does not wait for the lock of the connection phase"));
      // Past the deadline of the watchdog, which then found the connection still under way.
      Thread.sleep(Math.max(0, 2 * SHORT_CONNECTION_TIMEOUT_MS + 500
          - NANOSECONDS.toMillis(System.nanoTime() - beforeTheStop)));
      assertFalse(watchdogRuns(), "the watchdog of the stop did not act at its deadline");
      assertTrue(isIn(listener, ReplicationBroker.class, "connectAsDataServer"),
          "the connection ended before the watchdog of the stop acted: the case tests nothing");

      stop.join(stopBoundMS());
      assertFalse(stop.isAlive(), "stop() did not end once the connection did");
      assertEquals(rs.sessionsStarted(), 2, "the connection under way did not open its session");
      rs.waitForTheSessionToEnd();
      assertEquals(rs.count(EntryMsg.class), 0,
          "the stop wrote to the session a connection opened once the stop had given up");
      listener.join(stopBoundMS());
      assertFalse(listener.isAlive(), "the listener did not end with the stop");
    }
    finally
    {
      for (SilentReplicationServer silent : silentServers)
      {
        silent.shutdown();
      }
      cleanUp(rs, stopping, listener);
    }
  }

  /**
   * A give-up for credit closes nothing: a replication server which then drops the session is
   * reported as one which did, as it is with no stop at all. Only a close this server makes - by the
   * watchdog of a stop, or by the heartbeat monitor - is not reported, as the close is marked begun
   * before the sockets are closed.
   */
  @Test(timeOut = 120_000)
  public void aDropByTheReplicationServerAfterAGiveUpIsStillReported() throws Exception
  {
    final WithholdingReplicationServer rs = new WithholdingReplicationServer();
    rs.start();
    final CSNGenerator csns = new CSNGenerator(OTHER_DS_ID, 0);
    final AtomicReference<ReplicationBroker> brokerRef = new AtomicReference<>();
    final AtomicReference<Thread> listenerRef = new AtomicReference<>();
    final AtomicInteger reportedInTheStop = new AtomicInteger(-1);
    final ReplicationBroker broker = new ReplicationBroker(new DummyReplicationDomain(getGenerationId(baseDN))
    {
      @Override
      public void publishReplicaOfflineMsg()
      {
        // No credit: the change is given up at the credit deadline of the stop.
        brokerRef.get().publish(newDeleteMsg(csns));
        try
        {
          rs.dropTheSession();
          // The listener reports what ended its read before it waits for the end of the stop.
          newTimer().repeatUntilSuccess(() -> assertTrue(waitsIn(listenerRef.get(), ReplicationBroker.class, "reStart"),
              "the listener does not wait for the end of the stop"));
        }
        catch (Exception e)
        {
          throw new IllegalStateException(e);
        }
        reportedInTheStop.set(badDisconnectionsLogged());
      }
    }, new ServerState(), newFakeCfg(baseDN, OTHER_DS_ID, rs.getPort()), getReplSessionSecurity());
    brokerRef.set(broker);
    Thread listener = null;
    Thread stopping = null;
    try
    {
      connect(broker, 0);
      listener = startThread("listener", () -> {
        try
        {
          while (broker.receive() != null)
          {
            // What the replication server sends needs no answer here.
          }
        }
        catch (SocketTimeoutException e)
        {
          throw new IllegalStateException(e);
        }
      });
      listenerRef.set(listener);
      // The only credit of the window goes to a first change.
      broker.publish(newDeleteMsg(csns));
      rs.waitForUpdates(1);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();

      stopping = startThread("stop", broker::stop);
      stopping.join(stopBoundMS());
      assertFalse(stopping.isAlive(), "stop() did not end");
      assertTrue(reportedInTheStop.get() > 0,
          "the replication server dropped the session after a give-up for credit, and nothing reported it");
      assertEquals(rs.count(LDAPUpdateMsg.class), 1, "the change was written with no credit");
    }
    finally
    {
      cleanUp(rs, stopping, listener);
    }
  }

  /**
   * A replication server which is gone without a reset reads nothing, sends no heartbeat and answers
   * no connection. Once a write to it blocks, the heartbeat monitor still has to close the session,
   * not wait for that write, which nothing but TCP would end: the change then ends, and once the
   * replication server is back, the domain connects again and its changes reach it, the one cut
   * short included.
   */
  @Test(timeOut = 120_000)
  public void theHeartbeatMonitorEndsAWriteToAReplicationServerWhichIsGone() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);
    final WithholdingReplicationServer rs = new WithholdingReplicationServer(LARGE_WINDOW);
    final AtomicBoolean keepWriting = new AtomicBoolean(true);
    final AtomicInteger changes = new AtomicInteger();
    final AtomicReference<ResultCode> failed = new AtomicReference<>();
    Thread writer = null;
    try
    {
      final LDAPReplicationDomain domain = startDomain(rs, HEARTBEAT_INTERVAL_MS);
      TestCaseUtils.ERROR_TEXT_WRITER.clear();
      rs.goAway();
      final long away = System.nanoTime();
      writer = startLargeChanges(keepWriting, changes, failed);
      waitForAWriteToBlock(writer, changes);
      keepWriting.set(false);
      final long blocked = System.nanoTime();

      // The monitor gives up two or three intervals after the last heartbeat.
      writer.join(4 * HEARTBEAT_INTERVAL_MS + 10000);
      final long released = System.nanoTime();
      assertFalse(writer.isAlive(), "the write still blocks once the heartbeats of the replication server stopped");
      assertTrue(NANOSECONDS.toMillis(released - away) >= 2 * HEARTBEAT_INTERVAL_MS, "the write of change "
          + changes.get() + ", blocked " + NANOSECONDS.toMillis(blocked - away) + " ms after the replication server went away, ended "
          + NANOSECONDS.toMillis(released - away) + " ms after it: before the heartbeat monitor could give the session up");
      assertNull(failed.get(), "a local change failed");

      rs.resumeReading();
      waitForConnection(domain);
      final CSN last = domain.getServerState().getCSN(DS_ID);
      assertNotNull(last, "the domain has no change of its own in its state");
      newTimer().repeatUntilSuccess(() -> assertTrue(rs.receivedUpdateCSNs().contains(last),
          "the change cut short, " + last + ", did not reach the replication server: it has "
          + rs.receivedUpdateCSNs()));
      assertTrue(rs.sessionsStarted() > 1, "the domain kept the session whose heartbeats stopped");
      // The listener connected again: it is past what it reports of the read the close ended.
      assertEquals(badDisconnectionsLogged(), 0,
          "the close of the session by the heartbeat monitor was reported as one by the replication server");
    }
    finally
    {
      cleanUp(rs, writer);
    }
  }

  private LDAPReplicationDomain startDomain(WithholdingReplicationServer rs) throws Exception
  {
    // Heartbeats far apart: the session stays up whatever the replication server reads.
    return startDomain(rs, 100000);
  }

  private LDAPReplicationDomain startDomain(WithholdingReplicationServer rs, long heartbeatIntervalMS)
      throws Exception
  {
    rs.start();
    final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID,
        new TreeSet<>(Collections.singleton("127.0.0.1:" + rs.getPort())));
    conf.setHeartbeatInterval(heartbeatIntervalMS);
    final LDAPReplicationDomain domain = MultimasterReplication.createNewDomain(conf);
    domain.start();
    waitForConnection(domain);
    return domain;
  }

  private void waitForConnection(final LDAPReplicationDomain domain) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "the domain is not connected"));
  }

  /**
   * Waits for the thread which publishes the changes the replication server misses to end: until
   * then the domain does not push its own changes, but leaves them to that thread.
   */
  private static void waitForTheRecoveryToEnd() throws Exception
  {
    final String name = "Replica DS(" + DS_ID + ") missing change publisher";
    newTimer().repeatUntilSuccess(() -> {
      for (Thread thread : Thread.getAllStackTraces().keySet())
      {
        assertFalse(thread.isAlive() && thread.getName().startsWith(name), "the recovery still runs");
      }
    });
  }

  /** Sets the connection timeout of the replication provider, which the stop of a broker reads. */
  private static void setConnectionTimeout(long timeoutMS)
  {
    assertEquals(getRootConnection().processModify(
        modifyRequest(PROVIDER_DN, REPLACE, "ds-cfg-connection-timeout", timeoutMS + " ms")).getResultCode(),
        ResultCode.SUCCESS);
    assertEquals(MultimasterReplication.getConnectionTimeoutMS(), timeoutMS);
  }

  private DeleteMsg newDeleteMsg(CSNGenerator csns)
  {
    final CSN csn = csns.newCSN();
    return new DeleteMsg(DN.valueOf("cn=" + csn + "," + baseDN), csn, "entryUUID-" + csn);
  }

  /** No domain reads the session of the broker provided: the credit counts once the case reads it. */
  private static void readTheCredit(final ReplicationBroker broker) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> {
      try
      {
        broker.receive();
      }
      catch (SocketTimeoutException e)
      {
        // Nothing more to read.
      }
      assertEquals(broker.getCurrentSendWindow(), 1, "the credit the replication server gave");
    });
  }

  private static void await(final CountDownLatch latch)
  {
    try
    {
      assertTrue(latch.await(30, SECONDS), "a thread of the case did not get where the case waits for it");
    }
    catch (InterruptedException e)
    {
      throw new IllegalStateException(e);
    }
  }

  /** Waits for stop() to wait for the lock of the pending changes, which a local change holds. */
  private static void waitForTheOfflineMessageToWaitForTheLock(final Thread disabling) throws Exception
  {
    newTimer().repeatUntilSuccess(() -> assertTrue(disabling.getState() == Thread.State.BLOCKED
        && isIn(disabling, PendingChanges.class, "putReplicaOfflineMsg"),
        "stop() does not wait for the lock of the pending changes"));
  }

  /**
   * Returns how many times the error log says, since it was cleared, that the session was closed by
   * the replication server rather than by this server: what the listener reports when a read fails
   * on a session whose close it does not find begun.
   */
  private static int badDisconnectionsLogged()
  {
    int count = 0;
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    {
      count += record.contains("msgID=" + WARN_REPLICATION_SERVER_BADLY_DISCONNECTED.ordinal() + " ") ? 1 : 0;
    }
    return count;
  }

  /**
   * Starts a thread which makes changes of {@link #LARGE_MESSAGE_SIZE} until it is told to stop,
   * counting them, and keeping the first result which is no success.
   */
  private Thread startLargeChanges(final AtomicBoolean keepWriting, final AtomicInteger changes,
      final AtomicReference<ResultCode> failed)
  {
    return startThread("writer", () -> {
      while (keepWriting.get() && changes.get() < MAX_LARGE_MESSAGES)
      {
        final ResultCode result = replaceDescription(changes.get() + LARGE_VALUE);
        if (result != ResultCode.SUCCESS)
        {
          failed.compareAndSet(null, result);
        }
        changes.incrementAndGet();
      }
    });
  }

  /**
   * Waits for the writer to stay in a write to the session, the number of the changes it made
   * standing still: the socket buffers are full, and the replication server reads nothing.
   */
  private static void waitForAWriteToBlock(final Thread writer, final AtomicInteger changes) throws Exception
  {
    // Two seconds in a row, so that a pause of the collector in the middle of a write is not taken for it.
    new TestTimer.Builder().maxSleep(20, SECONDS).sleepTimes(50, MILLISECONDS).toTimer().repeatUntilSuccess(() -> {
      assertTrue(writer.isAlive(), "no write blocked after " + changes.get() + " changes");
      final int before = changes.get();
      for (int second = 0; second < 2; second++)
      {
        Thread.sleep(1000);
        assertTrue(changes.get() == before && isIn(writer, Session.class, "send"),
            "no write blocks after " + changes.get() + " changes");
      }
    });
  }

  /** Tells whether the watchdog of a stop of the broker of {@link #OTHER_DS_ID} runs. */
  private static boolean watchdogRuns()
  {
    final String name = "Replica DS(" + OTHER_DS_ID + ") watchdog of the stop";
    for (Thread thread : Thread.getAllStackTraces().keySet())
    {
      if (thread.isAlive() && thread.getName().startsWith(name))
      {
        return true;
      }
    }
    return false;
  }

  /** Tells whether the thread waits, for a lock or for a time, inside the method provided. */
  private static boolean waitsIn(final Thread thread, final Class<?> type, final String method)
  {
    final Thread.State state = thread.getState();
    return (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) && isIn(thread, type, method);
  }

  private static boolean isIn(final Thread thread, final Class<?> type, final String method)
  {
    for (StackTraceElement frame : thread.getStackTrace())
    {
      if (frame.getClassName().equals(type.getName()) && frame.getMethodName().equals(method))
      {
        return true;
      }
    }
    return false;
  }

  /**
   * Returns the CSN of the announcement that the replica of the domain went offline which the
   * shutdown of a collocated replication server would wait for the forward of, if there is one.
   */
  private CSN offlineAnnouncementOfTheReplica() throws Exception
  {
    final Field shutdownSync = MultimasterReplication.class.getDeclaredField("dsrsShutdownSync");
    shutdownSync.setAccessible(true);
    final Field announcements = DSRSShutdownSync.class.getDeclaredField("replicaOfflineMsgs");
    announcements.setAccessible(true);
    final Map<?, ?> byDomain = (Map<?, ?>) announcements.get(shutdownSync.get(null));
    final Map<?, ?> byReplica = (Map<?, ?>) byDomain.get(baseDN);
    final Object announcement = byReplica != null ? byReplica.get(DS_ID) : null;
    if (announcement == null)
    {
      return null;
    }
    final Field csn = announcement.getClass().getDeclaredField("csn");
    csn.setAccessible(true);
    return (CSN) csn.get(announcement);
  }

  private ResultCode replaceDescription(String value)
  {
    return getRootConnection().processModify(modifyRequest(baseDN, REPLACE, "description", value))
        .getResultCode();
  }

  /**
   * How long disabling the domain may take: the stop gives the replication server as long as a
   * connection to it to give credit, and as long again to read what was written, and the bound adds
   * a margin far beyond what the stop needs on top of that, and far below what a stop which waits
   * for the replication server would take: forever. It is also far below the 30 seconds of
   * connection timeout the suite runs with.
   */
  private static long stopBoundMS()
  {
    return 2 * MultimasterReplication.getConnectionTimeoutMS() + 10000;
  }

  private static Thread startThread(String name, Runnable runnable)
  {
    final Thread thread = new Thread(runnable, ReplicationBrokerStopTest.class.getSimpleName() + " " + name);
    thread.setDaemon(true);
    thread.start();
    return thread;
  }

  /**
   * Closes the replication server and then waits for the threads: on a broker which waits for credit
   * whatever happens, only the end of the session ends their wait - the broker finds no replication
   * server to connect to again and gives the publish up.
   */
  private void cleanUp(WithholdingReplicationServer rs, Thread... threads) throws Exception
  {
    rs.shutdown();
    for (Thread thread : threads)
    {
      if (thread != null)
      {
        thread.join(30000);
      }
    }
    // A no-op when no domain was registered.
    MultimasterReplication.deleteDomain(baseDN);
  }

  private static TestTimer newTimer()
  {
    return new TestTimer.Builder().maxSleep(10, SECONDS).sleepTimes(50, MILLISECONDS).toTimer();
  }

  /**
   * A replication server which keeps the session of the domain up and reads every message of it,
   * but gives no credit beyond the window it opens the session with - of one update, unless the
   * test opens a larger one - unless the test asks it to. Like a replication server, it reports the
   * changes it received as its state: a domain which finds itself ahead of it republishes what it
   * misses. It sends the heartbeats the domain asks for, stops reading its sessions - or goes away
   * altogether, heartbeats included - when the test asks it to, and serves each connection on a
   * thread of its own, as a replication server does: a session it does not read does not keep the
   * next connection waiting.
   */
  private static final class WithholdingReplicationServer extends Thread
  {
    private static final int RS_ID = 2;

    private final int window;
    private final ServerSocket listenSocket;
    /** The latest session started, until it ends. */
    private final AtomicReference<Session> session = new AtomicReference<>();
    /** Every session being served, the handshakes included, for the shutdown to close. */
    private final List<Session> sessions = new CopyOnWriteArrayList<>();
    private final List<Thread> servers = new CopyOnWriteArrayList<>();
    private final AtomicInteger sessionsStarted = new AtomicInteger();
    /** The kinds of the messages received on the sessions, in the order they came. */
    private final List<Class<?>> received = new CopyOnWriteArrayList<>();
    /** The CSNs of the changes received on the sessions, in the order they came. */
    private final List<CSN> updateCSNs = new CopyOnWriteArrayList<>();
    private final ServerState state = new ServerState();
    /** Guards {@link #reading} and the changes of {@link #away}. */
    private final Object readingLock = new Object();
    /** Whether the sessions read what the domain writes: each reads one message more once it is not. */
    private boolean reading = true;
    /** Whether this server is away: it then sends no heartbeat, and answers no connection either. */
    private volatile boolean away;

    WithholdingReplicationServer() throws IOException
    {
      this(1);
    }

    WithholdingReplicationServer(int window) throws IOException
    {
      super(ReplicationBrokerStopTest.class.getSimpleName() + " replication server");
      setDaemon(true);
      this.window = window;
      listenSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
    }

    int getPort()
    {
      return listenSocket.getLocalPort();
    }

    @Override
    public void run()
    {
      while (!listenSocket.isClosed())
      {
        final Socket socket;
        try
        {
          socket = listenSocket.accept();
        }
        catch (IOException e)
        {
          // The test closed this server.
          continue;
        }
        final Thread server = new Thread(() -> serve(socket), getName() + " session");
        server.setDaemon(true);
        servers.add(server);
        server.start();
      }
    }

    /**
     * Answers both phases of the handshake: the first connection of the domain only asks for the
     * state of this server and ends with a StopMsg, the one it keeps goes on to start a session.
     */
    private void serve(Socket socket)
    {
      Session s = null;
      try
      {
        // A server which is away answers no connection: the connection waits for it to come back.
        waitToBeBack();
        s = new ReplSessionSecurity().createServerSession(socket, MultimasterReplication.getConnectionTimeoutMS());
        sessions.add(s);
        final ServerStartMsg start = (ServerStartMsg) s.receive();
        final String url = "127.0.0.1:" + getPort();
        s.publish(new ReplServerStartMsg(RS_ID, url, start.getBaseDN(), window, state.duplicate(),
            start.getGenerationId(), start.getSSLEncryption(), start.getGroupId(), 5000));
        if (!start.getSSLEncryption())
        {
          s.stopEncryption();
        }
        if (!(s.receive() instanceof StartSessionMsg))
        {
          return;
        }
        sessionsStarted.incrementAndGet();
        s.publish(new TopologyMsg(new ArrayList<DSInfo>(), Collections.singletonList(
            new RSInfo(RS_ID, url, start.getGenerationId(), start.getGroupId(), 1))));
        s.setSoTimeout(0);
        session.set(s);
        startHeartbeats(s, start.getHeartbeatInterval());
        while (true)
        {
          waitToRead();
          final ReplicationMsg msg = s.receive();
          if (msg instanceof LDAPUpdateMsg)
          {
            final CSN csn = ((LDAPUpdateMsg) msg).getCSN();
            updateCSNs.add(csn);
            state.update(csn);
          }
          received.add(msg.getClass());
        }
      }
      catch (Exception e)
      {
        // The domain closed the session, or the test closed this server.
      }
      finally
      {
        if (s != null)
        {
          session.compareAndSet(s, null);
          s.close();
          sessions.remove(s);
        }
      }
    }

    private void waitToRead() throws InterruptedException
    {
      synchronized (readingLock)
      {
        while (!reading)
        {
          readingLock.wait();
        }
      }
    }

    private void waitToBeBack() throws InterruptedException
    {
      synchronized (readingLock)
      {
        while (away)
        {
          readingLock.wait();
        }
      }
    }

    /**
     * Sends heartbeats on the session at the interval the domain asked for, as a replication server
     * does from a thread of its own, until the session ends - and none while this server is away.
     */
    private void startHeartbeats(final Session s, final long intervalMS)
    {
      if (intervalMS <= 0)
      {
        return;
      }
      final Thread heartbeats = new Thread(() -> {
        try
        {
          while (!s.closeInitiated())
          {
            Thread.sleep(intervalMS / 2);
            if (!away)
            {
              s.publish(new HeartbeatMsg());
            }
          }
        }
        catch (IOException | InterruptedException e)
        {
          // The session ended.
        }
      }, getName() + " heartbeats");
      heartbeats.setDaemon(true);
      heartbeats.start();
    }

    /** Stops reading the sessions, after the message each of them is reading; the heartbeats go on. */
    void stopReading()
    {
      synchronized (readingLock)
      {
        reading = false;
      }
    }

    /**
     * Goes away as a server whose host is down does, with no reset of its connections: it reads
     * nothing more, sends no heartbeat, and answers no new connection.
     */
    void goAway()
    {
      synchronized (readingLock)
      {
        away = true;
        reading = false;
      }
    }

    /** Reads the sessions again - and, if it was away, sends heartbeats and answers connections again. */
    void resumeReading()
    {
      synchronized (readingLock)
      {
        away = false;
        reading = true;
        readingLock.notifyAll();
      }
    }

    /**
     * Reads the sessions again once the domain closed its session while this server read nothing,
     * and ends that session at once. What the domain left queued at its end goes out once its TCP
     * finds the window open again: from the update the reads here send, or else from its next window
     * probe - and after a stall of seconds the probes are seconds apart, 13 s after a stall of 14 s
     * on Linux, longer than the wait for the end of the session. A heartbeat the other way has the
     * closed end answer with a reset at once: this server reads what reached it, and the read ends.
     */
    void resumeReadingTheSessionTheDomainClosed()
    {
      final Session closed = session.get();
      resumeReading();
      if (closed != null)
      {
        try
        {
          closed.publish(new HeartbeatMsg());
        }
        catch (IOException e)
        {
          // The reset came already.
        }
      }
    }

    /** Ends the latest session in order, with a StopMsg, as a replication server which stops does. */
    void closeTheSession()
    {
      session.get().close();
    }

    /** Drops the latest session with no StopMsg, as a replication server which fails does. */
    void dropTheSession()
    {
      session.get().abort(new IOException("the replication server drops the session"));
    }

    /** Returns how many sessions were started, the connections which only ask for the state left out. */
    int sessionsStarted()
    {
      return sessionsStarted.get();
    }

    int count(Class<? extends ReplicationMsg> kind)
    {
      int count = 0;
      for (Class<?> msg : received)
      {
        count += kind.isAssignableFrom(msg) ? 1 : 0;
      }
      return count;
    }

    /** Returns the kinds of the updates received, replica offline messages included, in order. */
    List<Class<?>> receivedUpdates()
    {
      final List<Class<?>> updates = new ArrayList<>();
      for (Class<?> msg : received)
      {
        if (LDAPUpdateMsg.class.isAssignableFrom(msg) || ReplicaOfflineMsg.class.isAssignableFrom(msg))
        {
          updates.add(msg);
        }
      }
      return updates;
    }

    List<CSN> receivedUpdateCSNs()
    {
      return new ArrayList<>(updateCSNs);
    }

    void giveCredit(int credit) throws IOException
    {
      session.get().publish(new WindowMsg(credit));
    }

    /** Gives credit unless the domain ended the session, before this reads it or while this writes. */
    void giveCreditIfTheSessionIsUp(int credit)
    {
      final Session s = session.get();
      if (s != null)
      {
        try
        {
          s.publish(new WindowMsg(credit));
        }
        catch (IOException e)
        {
          // The domain closed the session in between.
        }
      }
    }

    void waitForUpdates(final int count) throws Exception
    {
      newTimer().repeatUntilSuccess(
          () -> assertEquals(count(LDAPUpdateMsg.class), count, "updates the replication server received"));
    }

    void waitForWindowProbes(final int count) throws Exception
    {
      newTimer().repeatUntilSuccess(() -> assertTrue(count(WindowProbeMsg.class) >= count,
          "window probes the replication server received"));
    }

    /**
     * Waits for the session to end: what the domain wrote to it has all been read by then - or, once
     * the session was ended by a reset, what of it reached this server.
     */
    void waitForTheSessionToEnd() throws Exception
    {
      newTimer().repeatUntilSuccess(() -> assertNull(session.get(), "the session of the domain is still up"));
    }

    void shutdown() throws Exception
    {
      listenSocket.close();
      resumeReading();
      for (Session s : sessions)
      {
        s.close();
      }
      join(30000);
      for (Thread server : servers)
      {
        server.join(30000);
      }
    }
  }

  /**
   * A replication server which takes each connection and, once told to, answers nothing on it, as
   * one which hangs does: a connection to it waits for the connection timeout in its SSL
   * negotiation. Until then it closes each connection at once, which fails it at once.
   */
  private static final class SilentReplicationServer extends Thread
  {
    private final ServerSocket listenSocket;
    private final List<Socket> held = new CopyOnWriteArrayList<>();
    private volatile boolean silent;

    SilentReplicationServer() throws IOException
    {
      super(ReplicationBrokerStopTest.class.getSimpleName() + " silent replication server");
      setDaemon(true);
      listenSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
    }

    int getPort()
    {
      return listenSocket.getLocalPort();
    }

    @Override
    public void run()
    {
      while (!listenSocket.isClosed())
      {
        try
        {
          final Socket socket = listenSocket.accept();
          if (silent)
          {
            held.add(socket);
          }
          else
          {
            socket.close();
          }
        }
        catch (IOException e)
        {
          // The test closed this server.
        }
      }
    }

    void answerNothing()
    {
      silent = true;
    }

    void shutdown() throws Exception
    {
      listenSocket.close();
      for (Socket socket : held)
      {
        socket.close();
      }
      join(30000);
    }
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java

@@ -16,8 +16,10 @@
package org.opends.server.replication.protocol;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
@@ -26,6 +28,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
@@ -138,6 +141,128 @@
    }
  }

  /**
   * A session with no thread writes on the publishing thread, as the session of a directory server
   * does: a peer which reads nothing holds that write, and with it a close of the session, which
   * waits for the lock of the writes. An abort ends both, and writes nothing more: the peer finds
   * the message cut short and then the end of the stream, with no StopMsg.
   */
  @Test(timeOut = 60_000)
  public void anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt() throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      final AtomicReference<Throwable> endOfTheWrite = new AtomicReference<>();
      final Thread writer = start("writer", () -> {
        try
        {
          pair.publisher.publish(newBlockingMsg());
        }
        catch (Throwable e)
        {
          endOfTheWrite.set(e);
        }
      });
      pair.awaitBytesReachedThePeer();
      final Thread closer = start("closer", pair.publisher::close);
      awaitWaitingIn(closer, "close");

      final Thread aborter = start("aborter",
          () -> pair.publisher.abort(new IOException("the test gives the session up")));
      aborter.join(SOCKET_TIMEOUT_MS);
      writer.join(SOCKET_TIMEOUT_MS);
      closer.join(SOCKET_TIMEOUT_MS);
      assertThat(aborter.isAlive()).as("the abort waits for the write").isFalse();
      assertThat(writer.isAlive()).as("the write still blocks after the abort").isFalse();
      assertThat(endOfTheWrite.get()).as("how the write ended").isInstanceOf(IOException.class);
      assertThat(closer.isAlive()).as("the close still waits for the write after the abort").isFalse();
      assertThatThrownBy(pair.peer::receive)
          .as("the peer read a message after the one the abort cut short")
          .isInstanceOf(IOException.class);
    }
  }

  /**
   * The thread of a session can be the one held in a write, and any close joins that thread first:
   * the abort ends the write before it joins it.
   */
  @Test(timeOut = 60_000)
  public void anAbortEndsAWriteOfTheThreadOfTheSession() throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      pair.publisher.start();
      pair.publisher.waitForStartup();
      pair.publisher.publish(newBlockingMsg());
      pair.awaitBytesReachedThePeer();

      final Thread aborter = start("aborter",
          () -> pair.publisher.abort(new IOException("the test gives the session up")));
      aborter.join(SOCKET_TIMEOUT_MS);
      assertThat(aborter.isAlive()).as("the abort waits for the thread of the session, held in its write")
          .isFalse();
      assertThat(pair.publisher.isAlive()).as("the thread of the session outlived the abort").isFalse();
      assertThatThrownBy(pair.peer::receive)
          .as("the peer read a message after the one the abort cut short")
          .isInstanceOf(IOException.class);
    }
  }

  /**
   * An abort marks the close begun before it closes the sockets: a reader whose read the close of
   * the sockets ends then finds the close begun, and does not take it for a failure of its peer -
   * which the broker of a domain reports as a disconnection by its replication server.
   */
  @Test(timeOut = 60_000)
  public void anAbortMarksTheCloseBegunBeforeItClosesTheSockets() throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      pair.peer.abort(new IOException("the test gives the session up"));
      assertThat(pair.peerSocket.closeBegunWhenClosed())
          .as("whether the close of the session was marked begun when its socket was closed")
          .isTrue();
    }
  }

  private static Thread start(final String name, final Runnable runnable)
  {
    final Thread thread = new Thread(runnable, SessionTest.class.getSimpleName() + " " + name);
    thread.setDaemon(true);
    thread.start();
    return thread;
  }

  /** Waits for the thread to wait, for a lock or for a time, inside the method of a session. */
  private static void awaitWaitingIn(final Thread thread, final String method) throws Exception
  {
    final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS);
    while (!waitsIn(thread, method))
    {
      assertThat(System.nanoTime() < deadline)
          .as(thread.getName() + " does not wait in Session." + method + "()")
          .isTrue();
      Thread.sleep(10);
    }
  }

  private static boolean waitsIn(final Thread thread, final String method)
  {
    final Thread.State state = thread.getState();
    if (state != Thread.State.WAITING && state != Thread.State.TIMED_WAITING)
    {
      return false;
    }
    for (final StackTraceElement frame : thread.getStackTrace())
    {
      if (frame.getClassName().equals(Session.class.getName()) && frame.getMethodName().equals(method))
      {
        return true;
      }
    }
    return false;
  }

  private static EntryMsg newBlockingMsg()
  {
    return new EntryMsg(SENDER_ID, PEER_ID, new byte[BLOCKING_MESSAGE_SIZE], 1);
@@ -160,7 +285,7 @@
  {
    final ReplSessionSecurity security = getReplSessionSecurity();
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    final Socket peerSocket = new Socket();
    final ObservedSocket peerSocket = new ObservedSocket();
    Socket publisherSocket = null;
    Session publisher = null;
    boolean connected = false;
@@ -180,6 +305,7 @@
      publisher = security.createServerSession(publisherSocket, SOCKET_TIMEOUT_MS);
      assertThat(publisher).as("could not create the session under test").isNotNull();
      final Session peer = peerEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      peerSocket.observe(peer);

      publisher.publish(new HeartbeatMsg());
      assertThat(peer.receive()).isInstanceOf(HeartbeatMsg.class);
@@ -204,13 +330,45 @@
    }
  }

  /**
   * The socket of the peer, which records, when it is first closed, whether the close of the
   * session it carries was marked begun by then.
   */
  private static final class ObservedSocket extends Socket
  {
    private volatile Session session;
    private volatile Boolean closeBegunWhenClosed;

    void observe(final Session session)
    {
      this.session = session;
    }

    /** Returns null until the socket of a session observed is closed. */
    Boolean closeBegunWhenClosed()
    {
      return closeBegunWhenClosed;
    }

    @Override
    public void close() throws IOException
    {
      final Session observed = session;
      if (observed != null && closeBegunWhenClosed == null)
      {
        closeBegunWhenClosed = observed.closeInitiated();
      }
      super.close();
    }
  }

  private static final class SessionPair implements Closeable
  {
    private final Session publisher;
    private final Session peer;
    private final Socket peerSocket;
    private final ObservedSocket peerSocket;

    private SessionPair(Session publisher, Session peer, Socket peerSocket)
    private SessionPair(Session publisher, Session peer, ObservedSocket peerSocket)
    {
      this.publisher = publisher;
      this.peer = peer;