[#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it

[vharseko]

Fixes #1014

The bug

ReplicaOfflineMsg has no encoding below replication protocol V8
(ReplicaOfflineMsg.getBytes() returns null there), Session.publish() treats that null
as "nothing to send" and returns as if it had sent it, and ServerWriter reports the forward
regardless of the version negotiated with the peer. Nothing between the two looks at that
version, so the shutdown bookkeeping records a delivery which never reached a socket.

The phantom forward is not a race the peer might lose. The writer of a pre-V8 peer has no
serialization and no socket write to do for that message, so it is the fastest possible path to
replicaOfflineMsgForwarded().

What that costs changed with #947, which landed while this was open and which this is now
rebased onto. Before it, one forward ended the wait of the whole domain: with one pre-V8 peer in
the topology, the entry the shutdown watches was systematically cleared before the writer of a
V8 peer had drained its queue, stopAllServers(true) then cleared that queue, and the peer
which could have been told never was - the cost #917 describes, caused deterministically
rather than by scheduling. #947 records the recipients of the message when put() queues it
and waits for each of them, so for such a message the phantom forward now strikes off only the
peer which reported it - which is where a pre-V8 peer has to end up, but as a peer which was
told, by a writer which sent nothing. What is left of the cost:

• an announcement no recipient was recorded for - the ones the javadoc of
  awaitReplicaOfflineMsgsForwarded() leaves to the first forward: the message of a replica
  which picked a remote replication server, the late announcement of A ReplicaOfflineMsg still queued behind an uncommitted change is recorded as sent, and the shutdown then waits its whole grace period for nothing #918 - still ends the wait
  of the whole domain on the first forward, whoever reports it. The copy ReplicaCursor
  synthesizes from the offline CSN for a peer which is catching up is how the writer of a pre-V8
  peer gets to be that first forward.
• ReplicationServer.awaitReplicaOfflineMsgsForwarded() still counts a pre-V8 peer as somebody
  to wait for: in a topology whose only peer predates V8, such an announcement waits out the
  whole grace period, since nothing can ever report a forward there.

A peer which negotiated V2..V7 can be there: ProtocolVersion.getCompatibleVersion() clamps
rather than refuses, there is no minimum supported version, and on the accept side
replicationServerDomain.register(this) sits outside the > REPLICATION_PROTOCOL_V1 block, so
such a peer is registered in connectedRSs and served by a full ServerWriter.

One correction to the issue

The issue attributes the cost to the old peer ("a peer left unaware keeps the replica in
ChangeNumberIndexer.getOldestLastAliveCSN()"). A peer which cannot decode the message predates
it - V8 exists because of ReplicaOfflineMsg (OPENDJ-1453), and ChangeNumberIndexer arrived
with it - so it has no consumer for the announcement and loses nothing. Not telling it is a
property of the protocol, not a defect. What is a defect is calling that a forward: the peers
which do run a ChangeNumberIndexer are the ones which pay for it.

The fix

• ServerWriter.isUpdateMsgFiltered() filters an update the peer cannot decode, before it
  is published, and reports it with WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER once per copy it
  takes - there is genuinely nothing to send. Per copy, not per announcement: the changelog road
  hands the message to the writer again on every refill of the late queue, since the offline CSN
  never enters the state of the handler (MessageHandler.updateServerState() leaves it out by
  design, and ReplicaCursor synthesizes the message anew from the offline CSN each time the
  cursor is reopened). The record does not distinguish consumers, because only one kind of
  consumer can get here: since A ReplicaOfflineMsg sent to a directory server on the catch-up path takes a send-window permit the session never gets back #1029 a ReplicaOfflineMsg never reaches the writer of a
  directory server at all - put() does not queue it for one, and
  DataServerHandler.updateServerState() drops the copy the changelog cursor of a directory
  server which is catching up synthesizes - and ReplicaOfflineMsg is the only message gated on
  a protocol version. This is the single point every delivery path passes through: the in-memory queue filled by
  ReplicationServerDomain.put(), and the changelog cursor of a peer which is catching up,
  where ReplicaCursor synthesizes a ReplicaOfflineMsg from the offline CSN and which a
  filter in put() would not cover. On top of [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947 a drop here is what it already is for the
  generation id filters: the writer reports replicaOfflineMsgNotForwarded() for the peer, so a
  pre-V8 peer the message was queued for is struck off by its own writer as a peer which will
  not forward it, rather than credited with a forward.
• The check goes through ReplicationMsg.isEncodableFor() (default true, overridden by
  ReplicaOfflineMsg), the queryable form of the null return getBytes() already documents.
  A message version-gated in a future protocol version is then covered without a new
  instanceof - the trap here is in the extension mechanism, not only in this one message.
• ReplicationServer.awaitReplicaOfflineMsgsForwarded() no longer counts such a peer as
  somebody to wait for. Its check was !domain.getConnectedRSs().isEmpty(), a stand-in for
  "there is somebody to forward the message to"; without narrowing it, the fix would turn the
  grace period ends too early into the grace period is always exhausted for an announcement no
  recipient was recorded for, in a topology whose only peer predates V8. An announcement put()
  did record recipients for no longer needs this on top of [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947 - the give-up of the writer ends
  its wait - which is what its javadoc now says.

The hot path keeps one virtual call returning a constant for every other message.

Tests

ReplicationServerShutdownSyncTest:

• onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline drives a peer through the real
  handshake at a chosen protocol version and asserts, for V8 and for V7, both what the peer
  received and what the writer reported to DSRSShutdownSync - through the
  RecordingShutdownSync of [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947, per peer: the V8 row expects the peer in forwardedBy(), the
  V7 row expects it absent from forwardedBy() and present in gaveUpOn(), and both expect it
  in dispatchedTo(), so that a writer which was never handed the message cannot pass either
  row. The V8 row is the positive control, so the V7 row cannot pass vacuously. The change
  published after the announcement is the synchronization point - the writer of a peer takes
  from one queue in order, so a peer which received that change has already dealt with the
  announcement which precedes it, and the assertions on what the writer reported cannot race.
  Both rows also read the error log of the run, through the errorLogRecordsOf() of
  ReplicationTestCase: the V7 row expects exactly one record of
  WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER for the peer, published as a warning and naming
  this replication server, the domain, the change and the version it has no encoding for; the
  V8 row expects none, so a guard which reported every message would fail there. The records
  are picked by message id and peer rather than by their whole text - the address the writer
  logs is the one the replication server sees the peer on, which the test side does not hold.
  Both rows wait for the peer to be served from its queue before the announcement is published
  (MessageHandler.isFollowing(), the accessor A replica which reconnects after a restore can be declared up to date and never sent the change it was restored past (ReSyncTest.testResyncAfterRestore) #963 left for this): a peer which is still
  catching up reads the changelog too, where ReplicaCursor synthesizes a second copy of the
  same announcement, and how many copies its writer takes - hence how many drops it reports -
  would be a race.
• shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage pins the caller side check, on an
  announcement no recipient was recorded for.
• shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage pins the other side of that
  check, on a mixed topology: two raw sessions registered on one domain, one at V7 and one at the
  current version, and an announcement no recipient was recorded for. The shutdown must wait out
  the whole grace period, for the sake of the peer which can still be told. What it asserts is
  the wait and not which peer decided it: the iteration order of connectedRSs is not the order
  the two are registered in.

SynchronizationMsgTest.replicaOfflineMsgHasNoEncodingBelowV8 pins the protocol floor itself,
so that getBytes() and isEncodableFor() cannot drift apart.

Both new assertions were watched failing against the code they pin, again after the rebase
onto #947:

mutation	fails with
the version guard out of ServerWriter	onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline[7, false] - the writer of the peer speaking protocol version 7 reported a forward of a message which never reached it, expected empty but was [92]
the version aware check out of awaitReplicaOfflineMsgsForwarded()	shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage - the shutdown waited for a peer which cannot decode the message, 5018 ms

Under the first mutation the V8 row passed next to the failing V7 one, so the failure is the
missing guard and not the harness; each mutation was run on its own test only.

The assertions on the log added in this round were watched failing the same way, each mutation
on the parametrized case alone:

mutation	fails with
the logger.warn of the drop removed	[7, false] - the drop for the peer speaking protocol version 7 should be logged once, expected size 1 but was 0
the drop reported for every message the writer takes, ahead of the version check	both rows - [8, true] a message the peer speaking protocol version 8 did receive was reported as dropped, expecting empty but holding the announcement and the change which follows it; [7, false] expected size 1 but was 3
logger.warn demoted to logger.info	[7, false] - the drop should be reported as a warning naming this replication server, the domain, the change and the version it has no encoding for, the record holds severity=NOTICE
the id of this replication server and the id of the peer permuted in the call	[7, false] - should be logged once, expected size 1 but was 0: no record names the peer as the server the update was not sent to

Regression, before the rebase: ReplicationServerTest, AssuredReplicationServerTest,
ProtocolCompatibilityTest, ProtocolWindowTest, GenerationIdTest, StateMachineTest,
ReplicationServerFailoverTest, ReplicationServerDynamicConfTest, TopologyViewTest,
DSRSShutdownSyncTest, MonitorTest, ChangelogBackendTestCase,
ChangeNumberControlPluginTestCase - 496 tests. The only failure was MonitorTest.setUp
binding the administration connector to 0.0.0.0:65534, the port collision several server tests
in one JVM produce; the class passes on its own. After the rebase onto #947:
ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest 76/76. After the rebase onto
#935 and #976, whose only conflict was the ordinal of WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER
(328 now, at the end of replication.properties; the sources are byte for byte those of the head the
table above was measured on): ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest
78/78, the two rows added by #975 included. After the rebase onto #958, #964, #974, #968 and #977,
whose only conflict was again the tail of replication.properties - master's 322 to 324 are kept
ahead of 328, nothing is renumbered, and the sources are still byte for byte the same:
ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest 78/78. After the rebase onto
#978, #980, #981, #984 and #987, whose one conflict was FakePeerReplicationServer - #987 split its
handshake into the two phases handshaking() and completeHandshake() need and turned the
constructors into the factories connected(...), so the protocol version rides on that shape now:
connected(port, id, baseDN, generationId, windowSize, protocolVersion) is the factory the test
calls, the private constructor takes the version and announces it in the start message as before,
and handshaking() passes the current one; the guard and the check are still byte for byte the
same: ReplicationServerShutdownSyncTest 18/18, the two cases of #987 included, SynchronizationMsgTest
78/78 and DSRSShutdownSyncTest 29/29. After the first review round, whose changes are the trace
on the directory server road, the assertions on the log, the wait for the peer to follow its queue
and one server id: ReplicationServerShutdownSyncTest 18/18. After the rebase onto the current
master - 28 commits, #1029 among them, and the one conflict was again the tail of
replication.properties, where master's 329 is kept and 328 is still free - and after the second
review round, whose changes are the guard reporting the drop whoever the consumer is, the case on
a mixed topology and the javadoc of one helper: ReplicationServerShutdownSyncTest 19/19,
SynchronizationMsgTest 78/78 and the ReplicaOfflineMsgCatchUpTest of #1029 2/2.

The assertions added in the second round were watched failing as well, each mutation run on the
cases it is meant for:

mutation	fails with
hasPeerWhichCanReceiveReplicaOfflineMsgs() deciding on the first handler iterated (return isSupportedBy(...) in the loop body)	shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage - the shutdown stopped waiting although a connected peer can still be told that the replica went offline, 2006 ms against the 5000 ms of the grace period; shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage stays green, so it is the mixed topology which catches it
the logger.warn of the drop removed, now that the guard has one arm	onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline[7, false] - the drop for the peer speaking protocol version 7 should be logged once, expected size 1 but was 0

Left out

• No filter in ReplicationServerDomain.put(). It would save a queue slot and a send window
  permit, but it does not cover the changelog cursor path, and the permit it would save is leaked
  by every filtered message anyway - see below. The record put() keeps for a pre-V8 peer is
  not what makes the give-up of [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947 work: giveUpOn() on an id no recipient list holds is a
  no-op. What that record costs is a wait on the backlog of that peer - with a V8 peer which has
  forwarded and a pre-V8 peer which is behind, the shutdown waits until the writer of the
  pre-V8 peer reaches the message and gives up, up to the whole grace period. That is where the
  phantom forward was reported before this change - the writer takes from one queue in order
  either way - so nothing takes longer than it did. A filter in put() would shorten that wait,
  at the price of the version term living in a second place while the writer still has to hold
  it for the changelog road, and of the queue road - the road by which a test can hand the
  message to a writer at all - no longer reaching the guard.
• The send window permit of a dropped message is still lost. ServerHandler.take() acquires
  it before any filter runs, the only release is updateWindow() on what the peer acked, and a
  peer acks only what it received - so every filtered message costs the session a permit for the
  rest of it, and after windowSize such drops the writer sits in tryAcquire(500 ms) and the
  peer receives nothing until it reconnects. That is true of the generation id and status filters
  which were there before, and the drop this PR adds only moves one such loss one step earlier
  (at the base Session.publish() dropped the same message after the same permit). windowSize
  is what the peer advertised in its start message, 100000 by default, so it takes as many drops
  to reach the stall. Raised as A message the writer filters keeps the send-window permit it took #1080 rather than changed here.
• The directory server side is untouched. A ReplicaOfflineMsg the broker refuses is still recorded as sent: ReplicationDomain.publish() discards the outcome #949 covers broker.publish() returning false,
  but a broker whose replication server predates V8 returns true while the session drops the
  message, so that hole stays open after A ReplicaOfflineMsg the broker refuses is still recorded as sent: ReplicationDomain.publish() discards the outcome #949 too.
• MonitorMsg.getBytes() returns null when its encoding throws, and Session.publish()
  swallows that the same way. It is the only other message which can produce a null encoding,
  and it is a different defect.

[vharseko]

Rebased onto master now that #947 has landed, and the description above is updated to what is left of the cost once the forward is waited for per peer. Nothing new is proposed here; this only says what the conflict was and how it was taken.

The two guards are unchanged. What #947 changed is what they are worth: for an announcement put() recorded recipients for, the phantom forward of a pre-V8 peer no longer ends the wait of the whole domain, it only strikes that peer off - as told, by a writer which sent nothing. The writer guard now turns that drop into the give-up #947 already reports for the generation id filters (replicaOfflineMsgNotForwarded()), so the peer is struck off for the right reason. The caller side check is still what keeps a topology whose only peer predates V8 from spending the whole grace period on an announcement no recipient was recorded for - the ones #947's javadoc of awaitReplicaOfflineMsgsForwarded() leaves to the first forward.

The conflict was in three places:

• awaitReplicaOfflineMsgsForwarded() javadoc - both paragraphs kept, and [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947's list of what the check still covers now names the peer which cannot decode the message.
• replication.properties - WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_310 keeps its ordinal next to [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947's 319 and 320; nothing renumbered.
• ReplicationServerShutdownSyncTest - ForwardCountingShutdownSync is dropped for [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947's RecordingShutdownSync, which records the same thing per peer, so onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline now asserts forwardedBy() / gaveUpOn() / dispatchedTo() for the peer rather than a count: the V7 row expects the peer absent from forwardedBy() and present in gaveUpOn(). FakePeerReplicationServer gets the protocol version as a sixth argument after [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947's send window instead of a second five-argument constructor, receiveUntil() reports through failed(), and the server ids of the two tests move off the ones [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947 took (8231 → 8234, 8229 + version → 8240 + version). The comment on the writer guard is reworded for the per-peer wait.

Re-run after the rebase: ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest 76/76, and both mutations of the table re-done against master - the V7 row fails with Expecting empty but was: [92] without the writer guard, shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage with 5018 ms without the caller side check.

[vharseko]

Ordinal moved: WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER 310 → 328 (5f7e04d). No other change.

310 is claimed by #935 as well, under a different name (WARN_REPLICATION_SERVER_ACCEPT_ERROR), and that branch holds the whole block 310 to 314. It is approved, so it lands first and the number is not mine to keep. Of what is left, 315 to 318 are claimed by #988, #985, #981 and #958, and 321 to 327 by #964, #968, #977, #981 and #959. 328 is the first ordinal no open branch claims.

The block moved to the end of the file rather than being renumbered where it stood: it sat between 309 and 319, and 328 there would have broken the ascending tail. That is where #968 and #977 ended up after their own moves, so the three agree - a renumbered message follows WARN_REPLAY_DRAIN_INTERRUPTED_320.

Nothing else moves with it. ServerWriter.java:186 references the constant by name, which the ordinal is not part of, and the message is named in no test and in no .adoc. The ordinals of replication.properties stay unique, and the twelve places where the file is not ascending are the ones that were there before.

Not measured again: the tests pin the filter and the wait, neither of which reads the ordinal.

[vharseko]

@maximthomas rebased onto master once more, now that #935 has landed (13d57e0, with #959, #975 and #976 in between), and the description is updated to the numbers below. Nothing new is proposed here; this only says what the conflict was and how it was taken.

The conflict was in one place: replication.properties. #935 holds 310 to 314 at the very spot ca39ef5 had inserted WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_310. Master's block is kept where it is and the message sits at the end of the file as 328 - which is where 5f7e04d had already put it, so that commit had nothing left to do once the first one was rebased, and it is gone: the PR is one commit again (f9f77fe). git range-diff against the previous head shows the Java and the tests byte for byte the same; the whole interdiff is the ordinal.

328 is still the first ordinal no open branch claims: 326 and 327 landed with #959, 321 to 325 are held by #964, #968, #977 and #981, and no open branch names 328 - checked on every open PR's copy of the file, not on the list from this morning.

Re-run after the rebase: ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest 78/78 - the two rows above the 76 of the previous round are the tests #975 added to that class, which merged cleanly.

Not measured again: the two mutations of the table. The sources under test are those the table was measured on, and the ordinal is the only thing that moved, which no test reads.

[vharseko]

@maximthomas rebased onto master once more, now that #958, #964, #974, #968 and #977 have landed (eef0757, 99cca92 now), and the description is updated to the run below. Nothing new is proposed here; this only says what the conflict was and how it was taken.

The conflict was in one place: the tail of replication.properties. Master appended 322 to 324 (#968, #977) after NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327, at the very spot f9f77fe had appended WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328. Master's block is kept where it is and 328 follows it, still at the end of the file; nothing is renumbered, the ordinals stay unique, and 328 is still claimed by no other open branch - checked on every open PR's copy of the file. git range-diff against f9f77fe shows the Java and the tests byte for byte the same; the whole interdiff is the context of that one hunk.

None of the five touches what this changes. #964 is the closest, in MessageHandler: a handler the domain feeds goes back to the changelog when a change is missing from its queue - a directory server handler only, isFedByTheDomain() is false for a peer replication server - and whatever it re-reads passes through ServerWriter.isUpdateMsgFiltered() like everything else, which is where the guard sits.

Re-run after the rebase: ReplicationServerShutdownSyncTest 16/16 and SynchronizationMsgTest 78/78.

Not measured again: the two mutations of the table. The sources under test are those the table was measured on.

[vharseko]

@maximthomas rebased onto master once more, now that #978, #980, #981, #984 and #987 have landed (129fc4e, c0f796d now), and the description is updated to the run below. Nothing new is proposed here; this only says what the conflict was and how it was taken.

The conflict was in one place: FakePeerReplicationServer in ReplicationServerShutdownSyncTest. #987 split its handshake into the two phases its own cases need - handshaking() stops before the TopologyMsg, completeHandshake() sends it - and turned the constructors into the factories connected(...), at the very spot this PR added the six-argument constructor which takes the protocol version. The version rides on #987's shape now: connected(port, id, baseDN, generationId, windowSize, protocolVersion) is what onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline calls, the private constructor takes the version and announces it in the start message as before, and handshaking() passes the current one. completeHandshake() and this PR's receiveUntil() sit next to each other.

ReplicationServer, ServerWriter, ReplicationServerDomain and replication.properties merged on their own: #987 moved listenThread.interrupt() below awaitReplicaOfflineMsgsForwarded() in shutdown(), and the version-aware check this PR puts inside that method does not care where the interrupt sits. 328 is still claimed by no other open branch - checked on every open PR's copy of the file: 329 #1044, 330 #1045, 331 #1049. git range-diff against 99cca92 shows the guard and the check byte for byte the same; the whole interdiff is the helper and the one call site.

Re-run after the rebase: ReplicationServerShutdownSyncTest 18/18 - the two cases of #987 included - SynchronizationMsgTest 78/78, DSRSShutdownSyncTest 29/29.

Not measured again: the two mutations of the table. The sources under test are those the table was measured on.

[maximthomas]

praise: The guard sits where every delivery road converges, and it reports what happened instead of pretending.

• ServerWriter.isUpdateMsgFiltered() (ServerWriter.java:175-190) covers both the queue filled by put() and the ReplicaCursor road, and the drop lands as replicaOfflineMsgNotForwarded() — a give-up #947's per-peer record can strike off, not a forward.
• ReplicationMsg.isEncodableFor() is the queryable twin of the null in getBytes(), and SynchronizationMsgTest.replicaOfflineMsgHasNoEncodingBelowV8 pins that the two cannot drift.
• The V7 row of onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline runs the real handshake through FakePeerReplicationServer.connected(…, REPLICATION_PROTOCOL_V7) next to the V8 row as its positive control, and the DeleteMsg published after the announcement makes the per-peer assertions race-free.

---

issue (non-blocking): The version guard also warns for a pre-V8 directory server, and once per late-queue refill rather than once per message.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:175-190, opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java:533-551, :766-772

isEncodableFor() is checked ahead of the isDataServer() branch, so a directory server which negotiated V2..V7 and is catching up from the changelog reaches it: fillLateQueue() reopens getCursorFrom(serverState) on every refill, ReplicaCursor synthesizes the ReplicaOfflineMsg from the offline replica's CSN each time, and updateServerState() never records an offline CSN (by design), so the same CSN is re-served on every refill while the handler is not following. At BASE that copy died silently in Session.publish(); at HEAD each copy logs WARN 328 — "not sending update … to server " — for a peer which has no consumer for the message and for which put() already drops it without a word. Behaviour is unchanged (the DS road reports nothing to DSRSShutdownSync), so this is the log contract only; the description's "once per message" holds on the queue road, not on the cursor road it names as the reason the guard lives in the writer.

if (!updateMsg.isEncodableFor(handler.getProtocolVersion()))
{
  if (!handler.isDataServer())
  {
    logger.warn(WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER,
        handler.getReplicationServerId(), updateMsg.getCSN(), handler.getBaseDN(),
        handler.getServerId(), session.getReadableRemoteAddress(),
        handler.getProtocolVersion());
  }
  return true;
}

Or: keep the WARN for every peer and trace it for a data server; either way, "once per copy the writer sees" in the description.

---

suggestion (non-blocking): WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER and its six arguments are pinned by no test.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:186-189, opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java:321-343

The V7 row asserts forwardedBy(), gaveUpOn(), dispatchedTo() and the peer's received messages; nothing reads the error log (git grep at the head: one reference under src/main, none under src/test). Deleting the logger.warn, permuting its arguments, or moving it so it fires per refill leaves ReplicationServerShutdownSyncTest 18/18 green — survival by construction, so no mutant was run. The call is right today (the six placeholders match the six arguments) and the neighbouring WARN_IGNORING_UPDATE_TO_* logs are equally unpinned, so this matches the class; it is the operator-visible half of the fix which has no pin.

// onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline, around the publish and the drain
final AtomicReference<List<ReplicationMsg>> drained = new AtomicReference<>();
final List<String> records = errorLogRecordsOf(() -> {
  shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
  broker.publish(new ReplicaOfflineMsg(offlineCSN));
  broker.publish(new DeleteMsg(DN.valueOf("uid=marker," + TEST_ROOT_DN_STRING),
      csns.newCSN(), "22222222-2222-2222-2222-222222222222"));
  drained.set(received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS));
  return null;
});
final List<ReplicationMsg> msgs = drained.get();
// ... the existing assertions on msgs and shutdownSync ...
if (!expectedToBeTold)
{
  assertThat(records)
      .filteredOn(r -> r.contains("severity=WARNING")
          && r.contains("msgID=" + WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER.ordinal())
          && r.contains("to server " + REMOTE_RS_ID))
      .as("the drop for the peer speaking protocol version %s is logged once", peerVersion)
      .hasSize(1);
}

Pin: ReplicationTestCase.errorLogRecordsOf(Callable<Void>) (ReplicationTestCase.java:963) is already in the superclass; hasSize(1) pins "once per message", and the severity=WARNING term is what a bare msgID+text filter would leave green under a warn→info edit.

---

suggestion (non-blocking): put() still records a pre-V8 replication server among the recipients the shutdown waits for.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java:411-417, :2390-2400, opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java:495-499

Recorded only — the description lists "No filter in put()" under Left out. Its stated reason, "so that the give-up has something to strike off", is not load-bearing: giveUpOn(id) on an id not in awaitedForwarders is awaited.remove(id) && … == false, a no-op, and the writer guard covers the ReplicaCursor road on its own. What the record costs: with a V8 peer which has forwarded and a V7 peer catching up, the collocated DS shutdown waits on the V7 backlog up to REPLICA_OFFLINE_GRACE_PERIOD before that writer reaches the message and gives up — the same length as the phantom forward at BASE, so not a regression. hasPeerWhichCanReceiveReplicaOfflineMsgs() already applies the term put() would need. Neither new case covers a recorded announcement with a behind V7 peer; both use an idle one. The send-window argument in Left out stands either way.

// pushReplicaOfflineMsgToReplicationServers()
if (!isDifferentGenerationId(rsHandler, offlineMsg)
    && offlineMsg.isEncodableFor(rsHandler.getProtocolVersion()))
{
  recipients.add(rsHandler);
  recipientIds.add(rsHandler.getServerId());
}

Or: leave it and say in Left out that the record is a wait on the peer's backlog, not a functional need.

---

nitpick (non-blocking): The test javadoc states the pre-#947 cost of a phantom forward.

opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java:262-264

"… or the shutdown stops waiting for the peers which do need the message" — on the #947 base a forward reported by a recorded peer strikes off that peer alone (PendingOfflineMsg.forwardedBy(id), DSRSShutdownSync.java:474-487); the whole domain's wait ends on the first forward only for an announcement no recipient was recorded for, the distinction the description and the ServerWriter comment (:178-181) both make. The row's assertions are the per-peer model and are right.

   * protocol it speaks - but the writer must not report that drop as a forward, or the shutdown
   * counts this peer as told by a writer which sent nothing - and, for an announcement no
   * recipient was recorded for, stops waiting for the peers which do need the message.

---

nitpick (non-blocking): Replication server id 8234 is used by two cases of the class.

opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java:1003, :370

shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage builds newReplicationServer(shutdownSync, "shutdownSyncOldPeerDb", 8234); thePeerWhoseHandshakeIsInFlight… already has 8234, every other case its own id (8221-8230). Cosmetic — the class runs sequentially and each case removes its server in finally — until a leaked server or a monitor entry named RS(8234) is attributed to the wrong case.

newReplicationServer(shutdownSync, "shutdownSyncOldPeerDb", 8231);

---

note (non-blocking): Not checked — nothing open.

• Execution evidence for the 18 cases and SynchronizationMsgTest at this head is CI run 35103446267 (ubuntu 11/17/21/25/26 green); no local run and no mutant was spent this round — the V7 row's gaveUpOn() == [92] cannot pass with the writer reading 8, which settles the V7 negotiation.
• Every message the RS emits to the V7 fake peer decodes at V7: ReplicaOfflineMsg is the only type gated at V8 (ReplicationMsg.java:85), and the guard withholds it.
• The DS-side twin (broker.publish() to a V7 RS, #949) is outside this PR, as Left out says.

[vharseko]

@maximthomas thanks - four taken, one recorded with the reason in the description corrected, and
one nitpick taken with a different id than the one suggested. The head is 5bfd31b.

The version guard also warns for a pre-V8 directory server, and once per late-queue refill.
Taken. The drop is now a trace on the directory server road and stays a warning for a peer
replication server: put() never queues the announcement for a directory server and says nothing
when it drops it, so the copies which reach the writer there are the ones ReplicaCursor
synthesizes while the handler is catching up, and a warning per copy reports a non-event. The
description says "once per copy it takes" now, and names what makes a copy come back - the offline
CSN never enters the state of the handler, so every refill of its late queue serves it again.

One correction to the reading, which does not change the outcome: the repetition is not specific to
the directory server road. MessageHandler serves a peer replication server from the same
fillLateQueue(), so a pre-V8 peer which is catching up sees the same synthesized copy on every
refill, and the warning it does keep is per copy as well. That is why the description now says
"once per copy" rather than restricting the claim to the road you named.

WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER is pinned by no test. Taken, in both rows: the V7 row
expects exactly one record of the message for the peer, published as a warning and naming this
replication server, the domain, the change and the version; the V8 row expects none, so a guard
which reported every message fails there too. The records are picked by message id and peer rather
than by their whole text - the address in the record is the one the replication server sees the
peer on, which the test side does not hold.

Pinning the count needed one thing your snippet did not have: the case now waits for the peer to be
served from its queue (MessageHandler.isFollowing()) before the announcement is published. A peer
which is still catching up reads the changelog too, where ReplicaCursor synthesizes a second copy
of the very announcement - the road the guard lives in the writer for - and hasSize(1) would have
been a race on it.

Four mutations, each run on the parametrized case alone:

mutation	fails with
the logger.warn of the drop removed	[7, false] - the drop for the peer speaking protocol version 7 should be logged once, expected size 1 but was 0
the drop reported for every message the writer takes, ahead of the version check	both rows - [8, true] a message the peer speaking protocol version 8 did receive was reported as dropped; [7, false] expected size 1 but was 3
logger.warn demoted to logger.info	[7, false] - should be reported as a warning …, the record holds severity=NOTICE
the id of this replication server and the id of the peer permuted in the call	[7, false] - should be logged once, expected size 1 but was 0

put() still records a pre-V8 replication server among the recipients. Recorded, and the reason
in Left out is corrected: you are right that giveUpOn() on an id no recipient list holds is a
no-op, so the record is not what makes the give-up work. What it costs is what you describe - a wait
on the backlog of that peer, up to the grace period, and not a regression, since the phantom forward
was reported at the same point of that same queue before this change. Left out says that now instead
of the load-bearing claim.

The filter is still left out, for two reasons: the version term would live in a second place while
the writer has to keep it for the changelog road anyway, and the queue road is how a test hands the
message to a writer at all - with put() filtering, the guard would be left covered only by the
changelog road, which no case reproduces.

The test javadoc states the pre-#947 cost. Taken, in your wording.

Replication server id 8234 is used by two cases. Taken, with 8236 rather than the 8231 of the
snippet: 8231 is shutdownSyncAbortedHandshakeDb (:773, from #987). 8236 to 8239 are free, and the
parametrized case keeps 8240 + version - now through a named constant, so that what the two rows
share is visible.

Run at this head: ReplicationServerShutdownSyncTest 18/18.

[maximthomas]

praise: The round-1 pin was taken as written and the delta says why each arm exists.

• ReplicationServerShutdownSyncTest.onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline:311-397 runs publish-and-drain inside errorLogRecordsOf() and pins the V7 record on hasSize(1) plus severity=WARNING, RS(id), the CSN, the domain and the version — an argument dropped or permuted fails there, and the V8 row's isEmpty() fails a guard which reports every message.
• waitForFollowing():1313-1331 polls MessageHandler.isFollowing() before the announcement, so how many copies the writer takes — and reports — is no longer a race with the changelog cursor.
• ServerWriter.isUpdateMsgFiltered():186-203 traces the directory-server drop and the comment names the mechanism (cursor road, offline CSN never in the handler state, once per refill).

---

issue (non-blocking): Every update the writer filters has already cost a send-window permit the peer never acks — pre-existing, on every filter arm.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java:991-994, :1059-1071, :1105, opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:175, :215-260

take() is getNextMessage() → acquirePermitInSendWindow() with no filter between; the only release is updateWindow()'s sendWindow.release(windowMsg.getNumAck()), and the peer acks only what it received (decAndCheckWindow:327-329, ReplicationBroker:2512). A message isUpdateMsgFiltered() drops — this arm or BAD_GEN_ID / FULL_UPDATE / WARN_IGNORING_UPDATE_TO_RS — never reaches the peer, so its permit is gone for the session; after windowSize (default 100) such drops the writer spins in tryAcquire(500 ms) and the peer receives nothing until reconnect. Not this PR's doing: at the base Session.publish() dropped the same message one step later (Session.java:308-314) after the same permit — the drop moved, the count did not. A follow-up issue; not run, the stall is derived from the three sites.

// ServerHandler
void releasePermitInSendWindow()
{
  sendWindow.release();
}
// ServerWriter.isUpdateMsgFiltered(), at every `return true`
handler.releasePermitInSendWindow();
return true;

Or: take the permit after the filter passes.

---

suggestion (non-blocking): On the peer-RS road WARN 328 fires once per copy, and a pre-V8 replication server catching up is handed the same announcement once per late-queue refill.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:204-211

Recorded only — the description states per-copy reporting as the choice. The reasoning the delta applies to the directory-server road holds for the cursor copies here too: ReplicationServerHandler reads through the same fillLateQueue() (MessageHandler.java:533-556, ≤100 msgs per refill, cursor reopened from serverState each time), ReplicaCursor.next():129-137 re-emits the message while the offline CSN is not covered, and only the queued copy — the one put() recorded a recipient for — strikes the peer off; every cursor copy's replicaOfflineMsgNotForwarded() is a no-op. A pre-OPENDJ-1453 peer N changes behind logs ~N/100 records for one announcement. The writer can tell the two copies apart: take() already reads isLastMessageFromLateQueue() (ServerHandler.java:992, protected in MessageHandler:277, same package).

if (handler.isDataServer() || handler.isLastMessageFromLateQueue())
{
  // a cursor copy: re-served once per refill, it reports nothing the queued copy did not
  if (logger.isTraceEnabled()) { logger.trace(...); }
}
else
{
  logger.warn(WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER, ...);
}

One corner: a queued copy evicted by the msgQueue overflow (MessageHandler.java:154) is then served by the cursor and would be traced; a once-per-(peer, offline CSN) memo in the handler avoids that at the cost of a field.

---

suggestion (non-blocking): The directory-server arm of the drop report — trace, not WARN — is entered by no test.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:186-203

Every pre-V8 fixture under src/test is a replication server: the FakePeerReplicationServer of the [V7,false] row (:263) and the raw session registered as a connected RS in shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage (:1073); the DS of the version rows is openReplicationSession(baseDN, LOCAL_DS_ID, …) at the current version. The mutant if (handler.isDataServer()) → if (false) — WARN 328 per refill for a pre-V8 DS, the round-1 shape this delta removes — leaves both rows green: the V7 row counts records naming to server 92 only. The inverse mutant is killed by hasSize(1); one side of the split is pinned. Deferred — the DS drop is a stated non-event and the fixture is the cost.

// a DS session hand-rolled at REPLICATION_PROTOCOL_V7 (the :1073 shape with a ServerStartMsg),
// connected with an empty state after an offline CSN is in the changelog, then:
final List<String> records = errorLogRecordsOf(() -> { /* publish the marker, drain */ return null; });
assertThat(records)
    .as("a drop for a directory server is traced, not reported")
    .noneMatch(r -> r.contains("msgID=" + WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER.ordinal())
        && r.contains("to server " + oldDsId));

---

suggestion (non-blocking): hasPeerWhichCanReceiveReplicaOfflineMsgs() is pinned at the two extremes only; a mixed V7+V8 topology is exercised by no case.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java:2390-2400

The only pre-V8 peers under src/test are single (ReplicationServerShutdownSyncTest:263, :1069-1078); every multi-peer case (:664-666, :918-920) connects current-version peers. The mutant which decides on the first handler iterated — loop body if (isSupportedBy(v)) return true; → return isSupportedBy(v); — is equivalent on one peer, on all-V8 peers and on none, so a regression which skips the wait for a V8 peer because a V7 peer came first stays green. Deferred.

Pin: the :1069-1078 shape with two registered sessions, one at V7 and one at V8, on an announcement no recipient was recorded for — awaitReplicaOfflineMsgsForwarded() must wait; the mutant returns false when the V7 handler is iterated first and ends the wait at once. Register order is not iteration order (ConcurrentHashMap), so assert the wait, not which peer decided it.

---

nitpick (non-blocking): waitForFollowing() was inserted between the javadoc of waitForConnectedReplicationServer() and that method.

opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java:1300-1305, :1334

The 40-line handshake-window comment ("The registration is not the end of the handshake …") now ends at :1303 directly above waitForFollowing()'s own /**; javac attaches only the immediately preceding doc comment, so it dangles and waitForConnectedReplicationServer() is undocumented. Move the waitForFollowing() block above that comment or below waitForConnectedReplicationServer().

[vharseko]

@maximthomas two taken, one settled by a change which landed on master after this review, one
recorded with a correction, and the fifth is the issue Left out already names. The branch is
rebased onto the current master (the one conflict was the tail of replication.properties:
master's 329 is kept and this PR's 328 is still free). The head is 8b1c504.

Every update the writer filters has already cost a send-window permit. Confirmed, and it is
the item Left out has been carrying - your reading sharpens what it costs, so Left out says it
your way now: the permit is taken in ServerHandler.take() before any filter runs, the only
release is updateWindow() on what the peer acked, and a peer acks only what it received, so a
filtered message costs the session a permit for the rest of it. After windowSize such drops
the writer sits in tryAcquire(500 ms) and the peer receives nothing until it reconnects.

One correction to the size: windowSize here is what the peer advertised in its start message
(ReplicationServerHandler:84, DataServerHandler:374), and the default of window-size is
100000 for both a replication server and a domain - the 100 is the fixture of these tests. So
the stall is real but takes 100000 drops on a default deployment.

Not this PR either way - the drop moved one step, the count did not - and the arms which can hit
it are the pre-existing ones. Raised as #1080, with the fix you give: take the permit after the
filter passes, or release it on every return true.

On the peer-RS road WARN 328 fires once per copy. Recorded, and it stays per copy. The two
roads are not the same event: put() queues the announcement for a peer replication server, so
a drop there is a message this replication server undertook to deliver and did not - the record
is the only account of it. Silencing the cursor copies by isLastMessageFromLateQueue() would
also silence the corner you name yourself - a queued copy evicted by the msgQueue overflow is
served by the cursor afterwards, and that is the one case where the delivery really was lost and
nothing else says so. The repetition is bounded: the copies stop as soon as the peer follows its
queue, and a peer which never catches up has a bigger problem than its log.

The directory-server arm of the drop report is entered by no test. It is entered by nothing
at all, which is why the mutant survives - and that is now true by construction: #1029 landed on
master a few hours after this review and gave DataServerHandler its own
updateServerState(), which drops a message that does not contribute to the domain state
before getNextMessage() hands it to the writer. A ReplicaOfflineMsg therefore never reaches
the writer of a directory server on either road: put() does not queue it for one, and the copy
the changelog cursor synthesizes is dropped by the handler. ReplicaOfflineMsg is the only
message gated on a protocol version, so the arm had no way in left.

So the split is gone rather than pinned: the guard reports the drop whoever the consumer is, and
the comment says why a directory server does not get here. What is left is one arm, which the V7
row of onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline enters and the V8 row
keeps empty. A message gated on a version in a future protocol and carried to a directory server
would be reported there - and would deserve it: unlike the announcement, it would be a change
that server is not getting.

hasPeerWhichCanReceiveReplicaOfflineMsgs() is pinned at the two extremes only. Taken, in
the shape you give: shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage registers
two raw sessions on one domain, one at V7 and one at the current version, announces a replica
offline with no recipient recorded for it, and asserts the shutdown waits out the whole grace
period. It asserts the wait and not which peer decided it, for the reason you give - the
iteration order of connectedRSs is not the registration order.

waitForFollowing() was inserted between a javadoc and its method. Taken: the block moved
below waitForConnectedReplicationServer(), which owns that comment again.

Run at this head: ReplicationServerShutdownSyncTest 19/19, SynchronizationMsgTest 78/78 and
the ReplicaOfflineMsgCatchUpTest of #1029 2/2. Two mutations, each on the cases it is meant for:

mutation	fails with
hasPeerWhichCanReceiveReplicaOfflineMsgs() deciding on the first handler iterated	shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage - the shutdown stopped waiting although a connected peer can still be told, 2006 ms against a grace period of 5000 ms; shutdownIsNotDelayedWhenNoConnectedPeerCanDecodeTheMessage stays green next to it
the logger.warn of the drop removed, now that the guard has one arm	onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline[7, false] - the drop for the peer speaking protocol version 7 should be logged once, expected size 1 but was 0

[maximthomas]

praise: The delta takes the round-2 mixed-topology gift and removes an arm that could never be reached.

• ServerWriter.isUpdateMsgFiltered():186-197 has one report road now. The premise holds: DataServerHandler.updateServerState():322 returns false for a message that does not contribute to the domain state, and MessageHandler.getNextMessage() hands a message out only when that call returns true (:374, :405, :439). So a ReplicaOfflineMsg never reaches a directory server's writer.
• ReplicationServerShutdownSyncTest.shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage:1113-1155 puts a V7 peer and a current peer in one domain. It kills the first-handler-decides and all-must-decode mutants of ReplicationServerDomain.hasPeerWhichCanReceiveReplicaOfflineMsgs():2390. It passed locally at the head: tests="1", 5.4 s.