[feat][io] Add Aeron Archive-backed lossless source mode

[david-streamlio]

Motivation

The Aeron source added in #127 / #128 is at-most-once, and that is inherent to what it reads.
Plain Aeron is a transport: no persistence, no resumable position, nothing to replay. Messages
published while the connector is restarting are simply gone, and a subscriber that falls behind
past term rotation loses data with no recovery path.

For a bridge whose entire purpose is getting data into durable storage, "we lose whatever arrived
during a restart" is the limitation people ask about first. It is also the one thing that blocks
using the connector as a system of record rather than a best-effort tap.

Aeron Archive is the answer, and it already
exists: it records publications to durable storage, replays from any position, and supports
replay-merge — a late joiner replays from a recorded position and then transitions seamlessly
into the live stream. That last piece is what makes a lossless restart possible without the
connector permanently lagging behind live.

Depends on #128. This builds on the source connector and the AeronPoller interface
introduced there. It should land after it.

Goal

Add an archive mode to the existing AeronSource that replays from a recorded position,
checkpoints its progress, and merges into the live stream — so a restart resumes where it left off
instead of losing the gap.

Why on the existing connector rather than a new one

pulsar-io.yaml declares at most one sourceClass per NAR — true of every module in this repo.
So the alternatives are worse:

• A second class in the same module is not discoverable by name; every deploy would need
  --classname org.apache.pulsar.io.aeron.AeronArchiveSource.
• A separate aeron-archive module means a second NAR re-bundling Aeron and duplicating the media
  driver lifecycle, and forces users to choose a NAR rather than a config value.

The poll loop already sits behind the AeronPoller interface for exactly this, so archive mode is
a second implementation rather than a rework:

AeronPoller (interface)
  ├── AeronPollingRunner    // plain transport, at-most-once      (exists)
  └── ArchivePollingRunner  // replay-merge + checkpoint           (this issue)

The mode must be explicit, not a boolean

The two modes have materially different delivery semantics, and that difference must be visible
in configuration rather than buried in a flag. A useArchive: true/false boolean invites exactly
the failure where someone flips it for a test and silently loses the guarantee.

Proposal: a mode field taking transport (default, preserving current behaviour) or archive,
logged at open(), with archive-only fields rejected when mode: transport so a
half-configured setup fails loudly rather than quietly running lossy.

Open decision: where does the checkpoint live?

This is the question that needs settling before implementation, and it is what separates a
medium change from a large one. Three candidates, none obviously correct:

Option	Pros	Cons
sourceContext.putState() / getState()	Purpose-built; no extra topics	Requires state storage (BookKeeper table service) enabled, which not every deployment has
A dedicated cursor topic	Works in any deployment	More moving parts; the connector manages its own compaction/retention
Archive recording position + Pulsar deduplication	No checkpoint storage at all	Depends on broker dedup being enabled; see caveat below

The third is the most interesting, and the hook exists. Record.getRecordSequence() returns
Optional<Long>. If the record carries the Aeron Archive position as its sequence, and broker
deduplication is enabled on the destination topic, then duplicates produced by replaying after a
restart are discarded by the broker. That turns at-least-once into effectively-once without the
connector implementing dedup itself.

Caveat worth settling in review: Pulsar dedup keys on (producerName, sequenceId) and requires
sequence IDs to increase monotonically. Archive positions are monotonic within a recording, but
reset across recordings — so a recording change would need handling, either by refusing to cross
recordings or by composing recording id into the sequence.

Configuration (provisional — depends on the decision above)

Field	Type	Default	Notes
mode	String	transport	transport | archive
archiveControlRequestChannel	String	—	Archive control request channel; required in archive mode
archiveControlResponseChannel	String	—	Archive control response channel
recordingId	long	-1	-1 discovers the latest recording matching channel + streamId
startPosition	long	-1	-1 resumes from the checkpoint, or the recording start if none
replayChannel	String	—	Channel used for the replay leg of replay-merge

Deployment note for the issue: Aeron Archive is a separate service. ArchivingMediaDriver can host
it in-process for single-node use, but a real deployment points at an externally managed archive —
so the existing useEmbeddedMediaDriver handling needs an archive-aware equivalent.

Suggested staging

Splitting this keeps the first PR reviewable:

1. Replay from a configured position, no checkpointing. Proves the Archive connection, recording
   discovery and replay plumbing. Delivers value on its own — a bounded backfill from an existing
   recording into Pulsar.
2. Checkpointing plus replay-merge. The actual lossless story, including whichever checkpoint
   mechanism review settles on, and the transition into the live stream.

Testing

• Unit — archive config validation; that archive-only fields are rejected in transport mode;
  recording-id discovery logic.
• Integration — an in-JVM ArchivingMediaDriver over aeron:ipc: record a publication, replay
  it into the source, assert every message arrives with correct positions. Then the case that
  matters: kill the source mid-stream, restart it, and assert no gap — which is the whole point
  and is not testable today.
• Container — as with the existing tests, only the broker leg can be containerised; an Aeron
  client reaches its media driver through memory-mapped files.

Acceptance criteria

• mode: archive on the existing AeronSource, with ArchivePollingRunner behind AeronPoller
• Archive-only config rejected when mode: transport; active mode logged at open()
• Replay-merge: replay from the checkpoint, then transition into the live stream
• Restart test proving no gap across a mid-stream kill
• Delivery semantics documented per mode — the existing at-most-once wording must stay accurate
  for transport and not be quietly overwritten
• Passes spotlessJavaCheck; every new config field carries @FieldDoc (the doc generator
  throws otherwise)
• Aeron listed under Sources in the README remains accurate

[david-streamlio]

Decisions for stage 2

Checkpoint storage: source state. sourceContext.putState() / getState(), using the
configured stateStorageServiceUrl. The checkpoint stores (recordingId, position) rather than a
bare position — see rotation below.

Recording rotation: follow automatically. A publisher restart opens a new recording and resets
positions to zero, which is routine rather than exceptional. When the current recording is stopped
and fully consumed, the source moves to the next recording for the same channel and stream id.

Guarantee: at-least-once. A crash between publishing and checkpointing replays a small window,
so duplicates are possible. Effectively-once is available on top, but only if the user enables
broker deduplication on the destination topic — records already carry the archive position as
getRecordSequence(). That is not promised by the connector, because it depends on user-side
configuration the connector cannot verify.

Correction to the issue body

The table above lists "Archive recording position + Pulsar deduplication" as a third checkpoint
option. It is not one, and the table is misleading.

With no stored position, a restart must replay from the beginning of the recording. Deduplication
discards duplicate messages; it does not avoid the replay work. On a recording holding a
billion messages, every restart would replay a billion messages to keep the tail. Dedup is a safety
net that belongs alongside a checkpoint, not an alternative to one.

Design change: continuous replay rather than ReplayMerge

The issue proposes ReplayMerge for stage 2. I intend to use continuous bounded replay —
repeatedly replaying [checkpoint, current recorded position) and advancing the checkpoint —
instead. Reasons, in order of weight:

1. It cannot outrun durability. Continuous replay consumes only what the archive has already
   recorded. ReplayMerge ends up reading the live stream, which means the connector can publish
   to Pulsar data that is not yet durable in the archive. For a bridge whose entire purpose is
   durable capture, never getting ahead of the recording is the more defensible behaviour.
2. It works over IPC. ReplayMerge requires a multi-destination subscription
   (control-mode=manual) and a UDP live destination, so it cannot serve an aeron:ipc channel at
   all. Continuous replay is one code path for both.
3. The latency it costs is irrelevant here. ReplayMerge wins on steady-state latency, but the
   Pulsar write path is already hundreds of microseconds of fsync. The difference disappears into
   it.

ReplayMerge remains the right answer for a latency-sensitive consumer and is worth revisiting if
one appears. Flagging the deviation here rather than quietly changing it.

Testing note

No real Aeron Archive deployment is available to test against, so coverage is limited to the
in-process ArchivingMediaDriver. That does not exercise control-channel configuration against a
separately deployed archive, or reconnection to one. Worth stating plainly as a known gap rather
than implying the archive path is proven in a realistic deployment.

[david-streamlio]

Review retrospective, for whoever picks up the follow-on work

#134 went through three review rounds and produced twelve findings, all of which were real,
plus two further bugs that CI caught and local runs could not. That is an unusually high hit
rate, and the defects were not scattered — they clustered tightly enough to be worth recording,
because the clusters predict where the remaining work will go wrong too.

Where the defects actually were

Cluster	Findings	What they had in common
The asynchronous acknowledgement seam	4	Code assumed a record was safe at the moment it was handed downstream
Aeron API assumptions	4	A parameter or return value meant something other than it appeared to
Failures reinterpreted as empty state	3	An exception or absent value silently became "nothing to do"
Test fidelity	1 + the meta-issue below	The double was more reliable than the real thing

The first cluster produced every serious defect. In order:

1. Checkpointing at enqueue time. consume() only queues; the framework publishes and
   acknowledges later. Committing on enqueue meant a crash could resume past records that were
   never published.
2. Checkpointing only after a replay chunk. Acknowledgements arrive asynchronously, so once
   caught up there was no further chunk to trigger a write and the committed position went
   unpersisted until shutdown. (Caught by CI, not review.)
3. Rotating with records still in flight. Resetting the tracker discarded pending entries —
   and because positions restart per recording, a late acknowledgement for an old position could
   collide with a successor position and advance its watermark.
4. Treating a stalled watermark as sufficient on failure. It stopped the checkpoint advancing
   past a failed record, but nothing re-emitted it, so the message stayed absent from Pulsar while
   the source looked healthy.

The cascade is the important part

Those four were not independent. Each fix created the next. Introducing the commit watermark to
fix (1) is what made (2), (3) and (4) possible — before it there was no separate committed cursor
to leave unpersisted, no tracker state to discard at rotation, and no watermark to mistake for a
retry mechanism.

That is worth stating plainly: a correctness fix in this area does not converge on its own. Anyone
extending it should expect the same and budget review accordingly, rather than treating a green
suite after one fix as the end.

Why the tests could not see any of it

The harness modelled acknowledgement as synchronous, immediate and always successful. Records
were collected from read() and, in the first revision, never acknowledged at all. The
consequences were systematic rather than incidental:

• The window between enqueue and acknowledgement did not exist, so (1) was unreachable.
• Acknowledgements were never late, so (2) was unreachable.
• Rotation never happened with records in flight, so (3) was unreachable.
• Publishing never failed, so (4) was unreachable.

The same shape appeared in the state double: getState and putState were modelled while
deleteState was left as a no-op, which made every resetCheckpoint test pass vacuously — they
would have succeeded had the delete done nothing at all.

The general lesson: a test double that is faster and more reliable than the real thing deletes
exactly the states where the bugs live. Every defect in cluster one lived in a state the harness
could not represent. That is not a missing test case; it is a harness encoding the bug's premise.

What is still not covered

Stated plainly so a green suite is not mistaken for proof:

• Catalog paging with gaps spanning more than one page — the listRecordingsForUri fix is
  reasoned, not exercised.
• Recordings with a non-zero start position, in both the initial and rotation paths — same.
• A real Aeron Archive deployment. Everything runs against an in-process
  ArchivingMediaDriver, which does not exercise control-channel configuration against a
  separately deployed archive, or reconnection to one.

Deferred, and why

getRecordSequence() was removed, not fixed. Exposing the archive position as a dedup sequence
is unsafe across rotation: positions restart, so the same producer emits decreasing sequence ids
and Pulsar's deduplication discards the new recording as already-seen — an optional optimisation
that becomes active message loss.

A correct version needs a sequence that is monotonic across rotation and stable across replay
(the same message must get the same sequence, or dedup cannot work). A persisted per-source base,
bumped by the length of each completed recording, satisfies both and is the obvious next attempt —
but it adds a third field to the checkpoint and a new way to be subtly wrong, so it wants its own
change and its own review rather than being smuggled in.

Suggested next steps

1. Rotation-monotonic dedup sequence — the deferred item above, on its own.
2. A harness that can represent the bad states: delayed acknowledgements, out-of-order
   acknowledgements, failures, and rotation while records are in flight. Everything in cluster one
   would have been catchable with one.
3. Coverage for the three untested paths listed above.
4. A run against a real deployed archive before anyone describes this mode as
   production-proven.