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

### 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](https://aeron.io/docs/aeron-archive/overview/) 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`. 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

Start with the existing AeronSource and AeronPoller from #128, then review the proposed ArchivePollingRunner and the three checkpoint options before implementation. Run the planned archive configuration and in-JVM ArchivingMediaDriver tests; done means replay-merge, restart recovery without a gap, accurate mode documentation, and passing spotlessJavaCheck.

Written by the indexing model from the issue text.