[Bug] Geo-replication can silently drop a message when a head-of-queue send failure rewinds the cursor after later pipelined messages already persisted

[lhotari]

Summary

In geo-replication with the V2 source-position deduplication path, a specific interleaving on the single FIFO replication producer (GeoReplicationProducerImpl) can cause a message to be permanently lost on the destination while being mark-deleted on the source as if it had been replicated — i.e. silent data loss, with no error surfaced.

The trigger is the conjunction of (a) a per-message head-of-queue send failure that fails only the head pending op and does not close the connection, while (b) later, higher-source-position messages already pipelined behind it get persisted at the remote. The replicator then rewinds its cursor and re-sends the head entry; by then the remote dedup high-watermark has advanced past it, so the re-sent entry is classified as a duplicate, dropped, and acked back as success — which advances the source mark-delete past a message that was never persisted at the destination.

This is a pre-existing issue on master and is independent of #26106 (see Scope below).

Interaction timeline

Scenario: the replicator reads a batch and sends E3, E4, E5 (source positions L:3, L:4, L:5, same source ledger L) in order on the single FIFO producer; E3 is the head of pendingMessages. Time flows top‑to‑bottom; each column is an actor.

t	Source replicator (cursor / mark-delete)	Source producer GeoReplicationProducerImpl (FIFO pendingMessages)	Remote broker (publish + dedup watermark)	Remote ledger
1	read batch E3,E4,E5; tag MSG_PROP_REPL_SOURCE_POSITION	enqueue E3(head),E4,E5
2		send E3 then E4,E5 in order
3			E3: pre-dedup reject (ChecksumError), connection stays open	E3 not persisted
4			E4,E5: NotDup → persist; watermark → L:5	E4,E5 persisted
5		recoverChecksumError (local buffer corrupted): remove only head E3, fail its callback
6	sendComplete(E3, error) → doRewindCursor(false)
7	E4,E5 acked (Case‑3) → asyncDelete(E4), asyncDelete(E5); mark-delete stays before E3
8	reads resume; rewind re-reads only E3 (E4,E5 already deleted)	re-send E3' (L:3)
9			E3' L:3 ≤ watermark L:5 → Dup → drop; ack success (target -1/-1)	E3 still not persisted = GAP
10		E3' ack → ackReceivedReplicatedMsg Case‑1 (≤ lastPersisted) → sendComplete(null) = success
11	asyncDelete(E3); mark-delete advances past E3 = SILENT DATA LOSS

sequenceDiagram
    autonumber
    participant R as Source replicator<br/>(cursor / mark-delete)
    participant P as GeoReplicationProducerImpl<br/>(FIFO pendingMessages)
    participant B as Remote broker<br/>(publish plus dedup watermark)
    participant L as Remote ledger

    R->>P: read and tag E3(L:3), E4(L:4), E5(L:5), send in order
    P->>B: send E3 (head)
    B-->>P: ChecksumError (pre-dedup, transient), connection stays open
    P->>B: send E4, E5
    B->>L: persist E4, E5
    B-->>P: ack E4, E5 (persisted watermark becomes L:5)
    Note over P: recoverChecksumError (local buffer corrupted),<br/>remove ONLY head E3 and fail its callback
    P-->>R: sendComplete(E3, error), then doRewindCursor(false)
    R->>R: asyncDelete(E4), asyncDelete(E5),<br/>mark-delete stays before E3
    R->>P: rewind re-reads E3 only, re-send E3-retry (L:3)
    P->>B: send E3-retry (L:3)
    Note over B: L:3 at or below persisted watermark L:5, classified Dup
    B-->>P: ack success (Dup-drop, target -1/-1),<br/>E3 never persisted = GAP
    Note over P: ackReceivedReplicatedMsg Case-1 (pos at or below lastPersisted),<br/>sendComplete(null) means success
    P-->>R: success
    R->>R: asyncDelete(E3), mark-delete advances past E3<br/>= SILENT DATA LOSS

Loading

Impact / Severity

• Impact: high — silent, permanent, unrecoverable data loss in geo-replication for the affected message(s). The source considers the message replicated; nothing logs an error.
• Probability: low — the only path that completes the full silent-loss chain is rare (see Trigger conditions). The far more likely observable symptom of these head-only failure paths is a recoverable replication stall with a destination gap, not silent loss.
• Net operational risk: low-to-medium. Worth a defensive fix because the failure mode, when it does occur, is silent and unrecoverable.

This is a normal reliability/correctness bug, not a security vulnerability.

Root cause: step-by-step chain (verified against master)

Scenario: replicator reads a batch and sends entries E3, E4, E5 (source positions L:3, L:4, L:5, same source ledger L) in order on the single producer. E3 is the head of pendingMessages.

1. FIFO send + source-position tagging.
GeoPersistentReplicator.replicateEntries tags each message with MSG_PROP_REPL_SOURCE_POSITION = "<ledgerId>:<entryId>" and sends them in order, one PersistentReplicator.ProducerSendCallback per entry.

• pulsar-broker/.../persistent/GeoPersistentReplicator.java:293-303

Batching is disabled on the replicator producer, so each entry is its own OpSendMsg with an independent source position and checksum:

• pulsar-broker/.../persistent/AbstractReplicator.java:156 (enableBatching(false))

2. Head-only failure that leaves the connection OPEN.
ClientCnx.handleSendError routes exactly two server errors to per-message recovery without closing the connection; every other error hits the default branch which closes the connection (note the comment at line 880):

• pulsar-client/.../impl/ClientCnx.java:869-882 — ChecksumError -> recoverChecksumError, NotAllowedError -> recoverNotAllowedError, default -> connectionClosed(...).

recoverNotAllowedError removes only the head op and fails only its callback (no resend, no connection close), leaving E4/E5 in flight:

• pulsar-client/.../impl/ProducerImpl.java:1510-1527

recoverChecksumError, when the local buffer is corrupted (verifyLocalBufferIsNotCorrupted(op) == false), removes only the head op and fails its callback, then returns (no resend, no close):

• pulsar-client/.../impl/ProducerImpl.java:1465-1493
• Note the non-corrupted branch instead resends all pending via resendMessages(...) (ProducerImpl.java:1507), which produces no gap.

On the broker, these per-message errors are emitted before the dedup check, so the head entry never advances the dedup watermark while E4/E5 do:

• NotAllowedError: PersistentTopic.publishMessage rejects max-message-size at pulsar-broker/.../persistent/PersistentTopic.java:659-663 and max-delivery-delay at :664-671, both before messageDeduplication.isDuplicate(...) at :673-674.
• ChecksumError: Producer.checkAndStartPublish fails checksum verification and sends ServerError.ChecksumError pre-publish/pre-dedup (pulsar-broker/.../service/Producer.java, ~:259-266).

So E4/E5 proceed to dedup as NotDup (isDuplicateReplV2) and persist; the persisted watermark advances to (L,5).

3. Failure rewinds the cursor; success deletes the entry.
PersistentReplicator.ProducerSendCallback.sendComplete rewinds the cursor on any non-InvalidMessageException error, and on success deletes the entry:

• pulsar-broker/.../persistent/PersistentReplicator.java:447-494 — error branch: beforeTerminateOrCursorRewinding(Failed_Publishing) + doRewindCursor(false) (lines 462-463); success branch: cursor.asyncDelete(entry.getPosition(), ...) (line 472).
• NotAllowedException / ChecksumException are siblings of (not subclasses of) InvalidMessageException, so the rewind fires for both.

E4/E5 were individually deleted, but mark-delete stays before E3; the rewind re-reads only E3 (E4/E5 skipped as already-deleted) and re-sends it as E3'.

4. Remote drops the re-sent lower position as a duplicate.
MessageDeduplication.isDuplicateReplV2 keeps a single per-producer high-watermark (keys producerName + "_LID" / "_EID"), not a per-entry/gap record:

• pulsar-broker/.../persistent/MessageDeduplication.java:389-457. The Dup classification is a two-level check: the message must first fall at-or-below highestSequencedPushed (lines 411-414) and then at-or-below highestSequencedPersisted (lines 439-443) to return MessageDupStatus.Dup. E3' (L:3) satisfies both against the (L,5) watermark → Dup.
• The persisted watermark is set with an unconditional put (not a max), carrying no record of skipped/never-persisted lower positions: MessageDeduplication.java:531-553 (recordMessagePersistedRepl, lines 550-551).

PersistentTopic acks a Dup as success: publishContext.completed(null, -1, -1):

• pulsar-broker/.../persistent/PersistentTopic.java:679-683.

E3 is therefore never persisted at the destination = permanent gap. The dedup cannot distinguish "never-persisted lower position" from "already-persisted replay".

5. Source mark-deletes the lost message.
The success SendReceipt hits GeoReplicationProducerImpl.ackReceivedReplicatedMsg Case-1 ("pending send is repeated due to repl cursor rewind"), because the pending op's source position (L:3) is <= lastPersistedSource (L:5):

• pulsar-client/.../impl/GeoReplicationProducerImpl.java:114-141 (condition at 118-120; removeAndApplyCallback(op, ..., false) at 138).
• removeAndApplyCallback calls op.sendComplete(null) (success) regardless of whether the remote actually persisted: GeoReplicationProducerImpl.java:260-283.
• lastPersistedSource reached (L,5) via the expected Case-3 path from E4/E5: GeoReplicationProducerImpl.java:159-172 (lines 168-169).

op.sendComplete(null) flows through to the broker-side ProducerSendCallback.sendComplete(exception == null) → cursor.asyncDelete(E3) (step 3 success branch). The source mark-delete advances past the never-persisted E3 = silent data loss.

There is no guard that (a) checks whether later-pipelined messages already persisted before rewinding on a head failure, or (b) distinguishes a real persist from a dedup-drop on the success ack (the SendReceipt carries target ledgerId/entryId -1/-1, forwarded verbatim, so the source has no distinguishing signal either).

Trigger conditions / how rare

All of the following must hold:

1. Geo-replication with V2 replicated dedup active on the destination (publishContext.supportsReplDedupByLidAndEid(), dedup enabled).
2. A pipelined batch where the head entry fails with an error routed to head-only recovery without closing the connection — only ChecksumError -> recoverChecksumError or NotAllowedError -> recoverNotAllowedError.
3. Later entries (higher source positions, same source ledger) get persisted at the remote, advancing the persisted watermark past the head position.
4. The head failure is pre-dedup (so the head never records a watermark) and transient (so the re-read/re-sent head passes the same broker-side check on retry and reaches dedup).

Reachability of the two candidate errors differs sharply — this is the weakest link, flagged honestly:

• NotAllowedError does NOT produce silent loss. Its geo-reachable triggers (max message size at PersistentTopic.java:659-663, max delivery delay at :664-671) are evaluated before dedup and are deterministic message properties. The rewound/re-read entry re-fails the identical pre-dedup check, never reaches dedup, and is never classified Dup. The outcome is a replication stall/loop with a destination gap (recoverable; backlog grows; visible). Actual data loss there requires an operator to force-skip the stuck entry. (The other NotAllowedError cases in Producer.checkAndStartPublish require a shadow topic / message position and are not reachable for a normal geo-replicator.)
• ChecksumError is the only path that completes the full silent-loss chain, and it additionally requires the client's local in-memory buffer of the head op to be genuinely corrupted (recoverChecksumError corrupted branch, ProducerImpl.java:1474-1493). Wire-corruption checksum errors with an intact local buffer take the non-corrupted branch and resend all pending (resendMessages, ProducerImpl.java:1507), producing no gap. Because the corruption is transient, the re-read entry carries a fresh valid buffer, passes broker checksum verification, reaches dedup, and is dropped as Dup. This demands genuine memory/bit corruption hitting exactly the head op — extremely rare.

Net: the chain is code-valid end-to-end, but the only fully-completing trigger in practice is ChecksumError under local client-buffer corruption (or an operator force-skipping a stalled NotAllowedError entry). This analysis is static; it has not been reproduced at runtime.

Scope: rewind paths and relation to #26106

• The bug lives on master's existing rewind paths: the failed-publish rewind (PersistentReplicator.ProducerSendCallback.sendComplete error branch → doRewindCursor) and, more generally, any path that rewinds the cursor while later pipelined messages have already persisted.
• Independent of [fix][broker] Fix replication stall when a cursor rewind skips an in-flight read #26106. [fix][broker] Fix replication stall when a cursor rewind skips an in-flight read #26106 fixes a replication stall when a cursor rewind races an already-dispatched in-flight read; its "Note on semantics" only claims at-least-once delivery with bounded duplicates and does not address this head-only-failure/dedup-drop gap, nor does it introduce it. However, [fix][broker] Fix replication stall when a cursor rewind skips an in-flight read #26106 may extend the reach: by letting the previously-stalling in-flight-read sub-case resume reads instead of stalling, it adds another route into the same unconditional rewind-on-head-failure logic, which lacks a guard against later-already-persisted entries.
• The V2 source-position dedup mechanism was introduced in [fix][broker] Geo Replication lost messages or frequently fails due to Deduplication is not appropriate for Geo-Replication #23697.
• Closest existing reports cover different failures: [Bug] Geo-replication V2 dedup watermark can be lost after target topic reload #25861 / PR [fix][broker] Prevent duplicate geo-replicated messages after target topic reload #25860 describe the opposite symptom (V2 watermark loss after target topic reload / producer-inactivity cleanup causing duplicates); [fix][broker] Fix replication stall when a cursor rewind skips an in-flight read #26106 fixes a stall. No existing issue covers this head-only-failure silent-gap scenario.

Suggested fix direction

The core defect is that a cursor rewind on a head-only per-message failure can advance the source mark-delete past an entry that was never persisted at the destination, because (a) the source cannot distinguish a dedup-drop ack from a real persist, and (b) the remote dedup keeps only a monotonic high-watermark with no gap record. Possible directions (to be discussed on dev@):

• Avoid the unconditional rewind-on-head-failure when later entries in the same in-flight batch have already been persisted (i.e. do not re-send a position that would now be dropped as a duplicate by the remote). At minimum, surface/fail the replication rather than silently advancing the cursor.
• On the source side, distinguish a genuine persist from a dedup-drop ack before deleting the source entry. The SendReceipt for a dedup-dropped message carries target ledgerId/entryId -1/-1; the source could treat a Case-1 ("repeated due to cursor rewind") ack with a sentinel target position as "not confirmed persisted" and refuse to mark-delete past it, instead retrying/stalling visibly.
• For the head-only checksum recovery path, reconsider dropping the head when the local buffer is corrupted (it produces a silent gap); failing the producer/replicator visibly would be safer than a silent drop.

A deterministic test (hold E4/E5 persistence while forcing a transient head-only failure on E3, then assert E3 is present on the destination and not mark-deleted on the source) would convert this from static-confirmed to runtime-proven before/with a fix.

Affected branch

master (static code analysis; all file:line references above verified against current master). Not runtime-reproduced.

[hiSandog]

The described failure mode is serious enough that a deterministic broker test would be valuable. If the head-of-queue send failure can be injected while later pipelined sends succeed, the test can assert cursor advancement and replication state do not silently drop the failed message. That would cover the ordering invariant without relying on a flaky cross-region setup.

[lhotari]

The described failure mode is serious enough that a deterministic broker test would be valuable. If the head-of-queue send failure can be injected while later pipelined sends succeed, the test can assert cursor advancement and replication state do not silently drop the failed message. That would cover the ordering invariant without relying on a flaky cross-region setup.

I guess one of the challenges is how to distinguish a persisted corrupted message from a transiently corrupted message. When the message that has persisted is corrupted, it's will remain corrupted. However, it's possible that a bug or some other condition corrupts the message transiently and it would recover after a retry / broker restart (or other runtime component restart).

[lhotari]

btw. I added a comment on this PR about first having discussions about the possible solution instead of directly opening a PR: #26124 (review)