[feat][io] Aeron source: Archive-backed lossless mode by david-streamlio · Pull Request #134 · apache/pulsar-connectors

david-streamlio · 2026-08-12T17:44:37Z

Fixes #133

Motivation

The Aeron source added in #128 is at-most-once, and that is inherent to what it reads: plain
Aeron is a transport with no persistence and no resumable position. Anything published while the
connector is restarting is simply gone. For a bridge whose purpose is getting data into durable
storage, that is the limitation people ask about first, and it blocks using the connector as a
system of record rather than a best-effort tap.

This adds an archive mode that replays from an Aeron Archive recording and checkpoints its
position, so a restart resumes rather than losing the gap. Transport mode is unchanged and remains
the default.

Modifications

Archive mode is a second AeronPoller implementation rather than a rework — which is what that
interface was introduced for in #128:

AeronPoller
  ├── AeronPollingRunner    // live transport, at-most-once     (existing)
  └── ArchivePollingRunner  // replay + checkpoint              (this PR)

It lives on the existing connector rather than a new one because pulsar-io.yaml declares at
most one sourceClass per NAR — true of every module in this repo. A second class would not be
discoverable by name, and a second module would re-bundle Aeron.

The mode is explicit, not a boolean. mode: transport | archive, logged at open(), with
archive-only settings rejected in transport mode rather than ignored. The two modes have
materially different delivery guarantees, and a config that looks set up for lossless replay while
quietly running at-most-once is the worst outcome available here.

Checkpointing uses the function state store and records (recordingId, position), not a bare
position: archive positions are only monotonic within a recording, and a publisher restart opens
a new one numbered from zero. Recording rotation is followed automatically, since a publisher
restart is routine rather than exceptional. ArchiveCheckpoint.requireAvailable() fails at
open() when stateStorageServiceUrl is absent — without that check the connector looks healthy
while re-ingesting the whole recording after every restart.

Precedence, and the escape hatch. A stored checkpoint outranks a configured startPosition;
the other order would silently re-ingest for anyone who left a startPosition in their config.
resetCheckpoint is the explicit way to reprocess history. It is one-shot, and documented as such
in the @FieldDoc (so it reaches the generated config reference), in the class javadoc, and in a
WARN logged on every start — because left in place it re-ingests on every restart and presents
as a duplicate-message problem long before anyone suspects a config value. It deletes the stored
checkpoint rather than ignoring it, so removing the flag afterwards leaves a usable checkpoint
rather than a stale pre-reset one.

Delivery is at-least-once. The checkpoint is written after records go downstream, so a crash in
between replays that window; checkpointEveryRecords (default 1000) bounds it. Records carry the
archive position as getRecordSequence(), so enabling broker deduplication on the destination
topic yields effectively-once — but that depends on user-side configuration the connector cannot
verify, so it is documented rather than promised.

Deviation from the plan in #133, flagged rather than made quietly

The issue proposes ReplayMerge. This uses continuous bounded replay instead — repeatedly
replaying [checkpoint, current recorded position) and advancing. Reasons, in order of weight:

It cannot outrun durability. Only what the archive has already recorded is consumed, so the
connector never publishes to Pulsar something not yet durable in the archive.
It works over IPC. ReplayMerge requires a multi-destination subscription and a UDP live
destination, so it cannot serve an aeron:ipc channel at all.
The latency it costs is irrelevant here — Pulsar's write path already spends hundreds of
microseconds on fsync.

ReplayMerge remains the right answer for a latency-sensitive consumer and is worth revisiting if
one appears. Discussion is on the issue.

Two Aeron behaviours worth knowing, found by testing rather than by reading

findLastMatchingRecording matches sessionId literally. It does not treat
Aeron.NULL_VALUE as a wildcard, so it returns NULL_VALUE unless the caller already knows the
session id — which a connector configured with a channel and a stream id never does. Probed against
a live archive: NULL_VALUE returns -1 whether the recording is active or stopped, while the real
session id returns the recording. Discovery therefore uses listRecordingsForUri, which takes no
session id and filters server-side.

A recording still being written has no stop position — the archive reports NULL_POSITION. So
replaying up to getStopPosition() would replay nothing against a live recording, which is the
common case rather than an edge one. It falls back to getRecordingPosition().

Verifying this change

130 tests in the module, 49 new, all passing locally along with spotlessJavaCheck and the NAR
build. The connector doc generator was run to confirm the new @FieldDoc fields render.

Integration (13 tests) against a real in-process ArchivingMediaDriver: a restart that
resumes mid-stream and delivers only what arrived while the source was down; messages published
before the source existed; larger-than-MTU reassembly on replay; recording-id and archive
position on every record; both directions of the checkpoint-versus-startPosition precedence;
resetCheckpoint reprocessing and rewriting rather than merely skipping; and fail-fast when no
recording matches.
Config (33 tests) covering the mode switch, every archive setting required in archive mode,
and every archive setting rejected in transport mode.
Test containers now start with withFunctionsWorker(), which drops the container's default
--no-functions-worker -nss and brings up the stream storage service the state store needs. The
archive tests back their SourceContext with real in-memory state rather than a bare mock —
which would have swallowed every checkpoint and made the resume assertions vacuous.

Known gap, stated plainly: no Aeron Archive deployment was available to test against, so
coverage does not exercise control-channel configuration against a separately deployed archive, or
reconnection to one. The in-process driver proves the record/replay/checkpoint cycle, not a
realistic deployment topology.

Does this pull request potentially affect one of the following parts?

Adds io.aeron:aeron-archive (Apache-2.0, same version as the existing aeron-client /
aeron-driver) to the aeron module only. No existing default changes: mode defaults to
transport, which is exactly today's behaviour.

Deployment note for anyone using archive mode: it needs a running Aeron Archive, and
stateStorageServiceUrl configured on the cluster for checkpointing. Both fail loudly at open()
rather than degrading silently.

Documentation

doc-required
doc-not-needed

Config is documented via @FieldDoc on AeronSourceConfig, which the docs module generates the
config reference from; verified by running the generator. Delivery semantics per mode, the start
precedence, and the ReplayMerge rationale are in class javadoc. The connector is already listed
under Sources in the README.

Adds an archive mode to the Aeron source that replays from an Aeron Archive recording and checkpoints its position, so a restart resumes instead of losing whatever arrived while the connector was down. Transport mode is unchanged and remains the default. Fixes apache#133 Archive mode is a second AeronPoller implementation rather than a rework, which is what that interface was added for. It lives on the existing connector because pulsar-io.yaml declares at most one sourceClass per NAR, so a separate class would not be discoverable by name and a separate module would re-bundle Aeron. The mode is an explicit `mode: transport | archive`, not a boolean, and archive settings supplied in transport mode are rejected rather than ignored. The two modes have different delivery guarantees, and a config that looks set up for lossless replay while quietly running at-most-once is the worst outcome available. The active mode is logged at open() for the same reason. Checkpointing uses the function state store and records (recordingId, position), not a bare position: archive positions are only monotonic within a recording, and a publisher restart opens a new one numbered from zero. Rotation is followed automatically, since a publisher restart is routine. requireAvailable() fails at open() when stateStorageServiceUrl is absent, because otherwise the connector looks healthy while re-ingesting the whole recording on every restart. A stored checkpoint outranks a configured startPosition — the other order would silently re-ingest for anyone who left a startPosition in their config. resetCheckpoint is the explicit escape hatch for deliberately reprocessing history. It is a one-shot flag and documented as such in the FieldDoc, in the class javadoc, and in a WARN logged on every start, because left in place it re-ingests on every restart and presents as a duplicate-message problem long before anyone suspects a config value. It deletes the stored checkpoint rather than ignoring it, so removing the flag leaves a usable checkpoint behind rather than a stale pre-reset one. Continuous bounded replay rather than ReplayMerge, which deviates from the plan in apache#133 and is flagged there rather than changed quietly. It cannot outrun durability, since only what the archive has already recorded is consumed; it works over IPC, which ReplayMerge cannot serve at all as it needs a multi-destination subscription and a UDP live destination; and the steady-state latency it costs disappears into Pulsar's fsync. Records carry the archive position as getRecordSequence(), so users who enable broker deduplication on the destination topic get effectively-once. That is not promised by the connector, because it depends on user-side configuration the connector cannot verify. Delivery is documented as at-least-once, with checkpointEveryRecords bounding the duplicate window. Two Aeron behaviours found by testing rather than by reading: findLastMatchingRecording matches its sessionId argument literally and does not treat Aeron.NULL_VALUE as a wildcard, so it returns NULL_VALUE unless the caller already knows the session id — which a connector configured with a channel and a stream id never does. Verified against a live archive: NULL_VALUE returns -1 whether the recording is active or stopped, while the real session id returns the recording. Discovery therefore uses listRecordingsForUri, which takes no sessionId and filters server-side. A recording still being written has no stop position; the archive reports NULL_POSITION. Replaying to getStopPosition() would replay nothing at all against a live recording, which is the common case, so it falls back to getRecordingPosition(). Test containers now start with withFunctionsWorker(), which drops the container's default "--no-functions-worker -nss" and brings up the stream storage service the state store needs. The archive tests back their SourceContext with real in-memory state rather than a bare mock, which would have swallowed every checkpoint and made the resume assertions vacuous. Tests: 130 in the module, 49 new. Thirteen integration tests run against a real in-process ArchivingMediaDriver, including a restart that resumes mid-stream and delivers only what arrived while the source was down, messages published before the source existed, larger-than-MTU reassembly on replay, and both directions of the checkpoint-versus-startPosition precedence. Known gap: no Aeron Archive deployment was available to test against, so coverage does not exercise control-channel configuration against a separately deployed archive, or reconnection to one.

…he tenant CI failed in AeronSourceContainerTest.setUp with "Namespace not found" while all 130 tests passed locally. The setup waited for the "public" tenant and then immediately created a producer on persistent://public/default/..., but bootstrap creates the namespace after the tenant. Waiting on the tenant was never sufficient; enabling the functions worker in the previous commit shifted startup timing enough to expose it, and CI is slower than a laptop so it surfaced there first. Now waits for public/default to exist as well. Applied to both container tests, since both got the functions worker.

Copilot

Pull request overview

Adds an Aeron Archive-backed source mode with replay, checkpointing, recording rotation, and configuration validation.

Changes:

Adds archive replay and state-store checkpointing.
Extends source configuration, metadata, and dependencies.
Adds archive integration/config tests and updates container setup.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
`gradle/libs.versions.toml`	Registers Aeron Archive dependency.
`aeron/build.gradle.kts`	Bundles Aeron Archive.
`AeronSource.java`	Selects transport or archive mode.
`ArchivePollingRunner.java`	Implements archive replay and rotation.
`ArchiveCheckpoint.java`	Persists replay checkpoints.
`AeronSourceConfig.java`	Adds archive configuration and validation.
`AeronRecord.java`	Adds recording metadata and sequences.
`AeronArchiveSourceIntegrationTest.java`	Tests archive replay and checkpoints.
`AeronSourceConfigTest.java`	Tests archive configuration.
`AeronSourceContainerTest.java`	Enables state storage in containers.
`AeronSinkContainerTest.java`	Aligns container bootstrap setup.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Eight findings from review, all verified against the code and the Aeron/Pulsar APIs before changing anything. Two invalidated the advertised guarantee. 1. At-least-once was false as written. consume() only queues a record; the framework publishes and acks later, so checkpointing at enqueue time committed positions for records still in memory. A crash in that window resumed PAST records that were never published — silent loss, in the mode meant to prevent it. Replay cursor and commit watermark are now separate: CommitTracker turns out-of-order acks into a contiguous acknowledged prefix, and only that is checkpointed. A failed record stalls the watermark rather than being stepped over, so a restart replays it. 2. The archive position was exposed as getRecordSequence() for broker dedup, but positions restart with each recording. After a rotation the same producer would emit decreasing sequence ids and dedup would discard the new recording as already-seen — an optional optimisation turned into active message loss. The sequence is removed; the position remains available as metadata. Doing this safely needs a sequence monotonic across rotation, which is follow-up work. Noted in apache#133 as a hazard and then implemented anyway; caught in review. 3. listRecordingsForUri takes an inclusive recording ID, not an offset. Advancing by the match count re-read the same page whenever matching IDs had gaps, looping forever. Now steps past the highest ID each page returned. 4/5. A recording does not necessarily begin at position 0, and neither does a successor after rotation. Both now read the descriptor's start position rather than assuming zero, which would have failed the replay asynchronously. 6. An unexpected failure killed the poll thread while AeronSource stayed open and read() blocked forever, so a dead connector reported itself healthy. Now reports through SourceContext.fatal() so the runtime can restart the instance, which resumes from the checkpoint. 7. clear() swallowed delete failures, so a failed reset could resurrect the stale position. Deleting an absent key can legitimately throw, so the outcome is now verified by reading back rather than inferred from the exception, and a checkpoint that survives deletion fails open(). 8. The test stub implemented getState and putState but left deleteState as Mockito's no-op, so every reset test would have passed even if clear() deleted nothing. Now modelled. The tests could not have caught finding 1: the reader collected records without ever acknowledging them, modelling a world where publishing always succeeds instantly. It now acks as the framework does, and a test withholds acks to assert nothing is checkpointed while records are unpublished. Tests: 134 in the module, 4 new covering the unacknowledged-record case, the acknowledged-prefix watermark, reset failing loudly when deletion does not take, and reset actually deleting. Findings 3, 4 and 5 are fixed but not directly covered: they need a catalog with ID gaps beyond one page, and recordings with non-zero start positions, neither of which the in-process archive makes convenient to construct.

…er a chunk CI failed testResetActuallyDeletesTheStoredCheckpoint waiting for a checkpoint that never arrived. The test was fine; the loop was wrong. writeCheckpoint was only attempted after replaying a chunk. Acknowledgements arrive on framework threads, so the commit watermark advances asynchronously — and once the source has caught up there is no further chunk to trigger a write. If the acks landed a moment after that last attempt, the committed position was never persisted until shutdown. Locally the acks happened to land in time; CI's slower scheduling did not. The consequence in production is a stale checkpoint rather than lost data: a crash would re-replay from an older position than the source had actually committed. That is safe but wasteful, and on a recording that stops growing the checkpoint could stay stale indefinitely. The loop now attempts a checkpoint on every pass, including idle ones, and writeCheckpointIfAdvanced skips the write unless the watermark actually moved, so idle passes stay cheap. Verified by running the archive suite three times from clean rather than once — the original failure was timing-dependent and a single green pass proves little.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.

Suppressed comments (2)

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java:52

This documentation still promises a record sequence backed by archive position, but the implementation and testNoRecordSequenceIsExposed deliberately removed it because positions decrease across recording rotation. Document archive mode as at-least-once and state that positions are metadata only, otherwise users may enable broker deduplication expecting a guarantee the connector no longer provides.

 *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
 *       storage exists and data missed while the connector was down can be re-read. Records carry
 *       the archive position as their record sequence, which broker deduplication can use.

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:303

Calling writeCheckpointIfAdvanced() on every loop pass bypasses checkpointEveryRecords once replay has caught up. As acknowledgements arrive asynchronously, each newly advanced acknowledgement batch can trigger a remote state-store write even with the default interval of 1000, so the configured write-cost tradeoff is not honored. Track newly committed records (rather than emitted records) and apply the interval to normal writes, while retaining explicit final/rotation flushes.

                // Attempted on every pass, including idle ones. Acknowledgements arrive on
                // framework threads, so the watermark can advance long after the last record was
                // read — and once caught up there is no further chunk to trigger a write. Only
                // checkpointing after a chunk left the committed position unpersisted until
                // shutdown whenever the acks landed a moment too late.
                writeCheckpointIfAdvanced();

Three findings from the re-review, all in code the first review never saw — the CommitTracker work landed after it. All three verified before changing anything. Rotation could drop in-flight records. advanceRecordingIfFinished switched recordings as soon as the replay cursor reached the stop position, even with records still awaiting acknowledgement, and resetting the tracker discarded them. Worse, because positions restart in the successor, a late acknowledgement for an old position could collide with a successor position and advance ITS watermark, skipping unpublished successor records after a crash. Rotation now waits for the pending count to drain and persists the old recording's final watermark before switching. A record the framework never acknowledges holds rotation indefinitely, which is deliberate — advancing would skip unpublished data — so it warns on a throttled interval rather than stalling silently. read() turned every state-store exception into "no checkpoint". A transient read failure therefore restarted from configured history and silently re-ingested the recording, which is the exact failure this mode exists to prevent. It now returns empty only for a confirmed absent or truncated value and propagates store failures. clear() no longer mistakes "delete failed and verification also failed" for a successful reset; it fails and says the outcome is unknown. recordingStartPosition ignored listRecording's return value. A configured recordingId that does not exist produced a synthetic position 0, after which every position query answered NULL_POSITION and the loop idled forever while the source reported itself healthy and emitted nothing. It now fails at open(). Tests: 136 in the module, 2 new — a nonexistent recordingId failing fast, and an unreadable checkpoint store failing rather than re-ingesting. Archive suite run three times from clean, since the last two defects here were timing-dependent.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Suppressed comments (5)

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java:52

This documents a sequence ID that archive records deliberately no longer expose. ArchivePollingRunner and testNoRecordSequenceIsExposed explicitly keep getRecordSequence() empty because positions decrease across recording rotation. Leaving this claim tells users broker deduplication is safe when enabling it could instead drop data; document the actual at-least-once behavior and metadata properties.

 *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
 *       storage exists and data missed while the connector was down can be re-read. Records carry
 *       the archive position as their record sequence, which broker deduplication can use.

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:317

This unconditional idle-loop call bypasses checkpointEveryRecords: whenever acknowledgements arrive asynchronously after replay catches up, each watermark advance can cause another state-store write. Thus the default can write nearly once per record rather than once per 1000 records. Gate normal writes on acknowledged progress reaching the configured interval, while retaining forced flushes for rotation and shutdown.

                // Attempted on every pass, including idle ones. Acknowledgements arrive on
                // framework threads, so the watermark can advance long after the last record was
                // read — and once caught up there is no further chunk to trigger a write. Only
                // checkpointing after a chunk left the committed position unpersisted until
                // shutdown whenever the acks landed a moment too late.
                writeCheckpointIfAdvanced();

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:190

The checkpoint path does not verify that its recording still exists. If archive retention removed the checkpointed recording (or the archive was replaced), both position queries return NULL_POSITION; the loop then treats it as an active recording and idles forever while the source appears healthy. Validate the descriptor during open() just as the explicit recordingId path does, and fail rather than silently hanging when durable history is gone.

        if (stored.isPresent()) {
            currentRecordingId = stored.get().recordingId();
            currentPosition = stored.get().position();

aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java:472

This test does not exercise the advertised mid-recording restart: it waits until every record is delivered and acknowledged, closes cleanly, then recordMessages(second) creates a separate successor recording. The core resume case—checkpoint partway through an active recording, publish more while the source is down, then resume that same recording—remains untested. Stop after an acknowledged prefix while keeping the recording active, publish the remainder, and assert the restart emits only that remainder.

    public void testResumesFromCheckpointAfterRestart() throws Exception {
        // The property this whole mode exists for: a restart continues rather than re-ingesting.
        List<String> first = new ArrayList<>();
        for (int i = 0; i < 50; i++) {
            first.add("first-" + i);
        }
        recordMessages(first);

        Map<String, Object> config = archiveConfig();
        // Checkpoint aggressively so the restart has something recent to resume from.
        config.put("checkpointEveryRecords", 1);
        startSource(config);
        awaitRecords(first.size());

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java:260

recordingId and startPosition values below the documented -1 sentinel are accepted, but resumeOrStart() treats every negative value as unset via >= 0. For example, startPosition: -2 silently starts at the recording beginning instead of rejecting an invalid operator value. Validate both fields as >= -1 so mistyped positions are not ignored.

        if (isArchiveMode()) {
            requireArchiveField(archiveControlRequestChannel, "archiveControlRequestChannel");
            requireArchiveField(archiveControlResponseChannel, "archiveControlResponseChannel");
            requireArchiveField(replayChannel, "replayChannel");

Third-round review finding, against the current head this time. Stalling the commit watermark on a failed publish stopped the failure being checkpointed past, but nothing re-emitted the record: the replay cursor had already moved on, so the message stayed absent from Pulsar while the source went on looking healthy. Only an external restart would have retried it, and nothing triggered one. That is not at-least-once. A failed record now also fails the source via SourceContext.fatal(), so the runtime restarts the instance and the stalled watermark decides where replay resumes. Same approach as the Kinesis source, which does this for the same reason. CommitTracker keeps stalling the watermark and its javadoc now says plainly that this is only half the answer — it holds the position, the caller decides the policy. Tests: 137 in the module, 1 new asserting a failed publish reports fatal and leaves nothing checkpointed.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Suppressed comments (5)

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java:104

A truncated checkpoint is present but unreadable; treating it as absent starts from configured history and silently re-ingests the recording, exactly the failure this class otherwise avoids for state-read errors. Fail startup on a malformed checkpoint instead of returning empty.

        if (buffer.remaining() < SERIALIZED_BYTES) {
            LOG.warn("Ignoring a truncated archive checkpoint under {} ({} bytes, expected {})",
                    key, buffer.remaining(), SERIALIZED_BYTES);
            return Optional.empty();

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java:52

This documents a record sequence that archive mode deliberately does not expose: ArchivePollingRunner says no sequence is set, and testNoRecordSequenceIsExposed asserts it is empty. Leaving this claim directs operators to enable deduplication that cannot work. Document the at-least-once checkpoint semantics and position metadata instead.

 *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
 *       storage exists and data missed while the connector was down can be re-read. Records carry
 *       the archive position as their record sequence, which broker deduplication can use.

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java:304

Comparing primitives with their defaults cannot distinguish “unset” from an explicitly supplied default. For example, mode: transport with checkpointEveryRecords: 1000, resetCheckpoint: false, recordingId: -1, or startPosition: -1 passes validation, contrary to the stated guarantee that archive-only keys are rejected rather than ignored. Preserve key presence during deserialization (or use nullable fields and apply defaults after validation) so every supplied archive-only setting is detected.

        if (recordingId != -1L) {
            unexpected.add("recordingId");
        }
        if (startPosition != -1L) {
            unexpected.add("startPosition");
        }
        if (checkpointEveryRecords != DEFAULT_CHECKPOINT_EVERY_RECORDS) {
            unexpected.add("checkpointEveryRecords");
        }
        if (resetCheckpoint) {
            unexpected.add("resetCheckpoint");

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:195

A checkpoint’s recording ID is trusted without checking that the recording still exists. If archive retention removes that recording, both position queries return NULL_POSITION; the run loop then treats it as an active recording and idles forever while the source appears healthy. Validate the checkpoint recording descriptor during open() just as the explicit recordingId path does.

        if (stored.isPresent()) {
            currentRecordingId = stored.get().recordingId();
            currentPosition = stored.get().position();

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:322

This unconditional idle-pass write bypasses checkpointEveryRecords. In the steady-state trickle case, each acknowledgement advances the watermark and the next loop writes state, so the default of 1000 can still cause one state-store write per record. Count committed records since the last persisted watermark and gate ordinary writes on that count, while retaining forced flushes for rotation and shutdown.

                // Attempted on every pass, including idle ones. Acknowledgements arrive on
                // framework threads, so the watermark can advance long after the last record was
                // read — and once caught up there is no further chunk to trigger a write. Only
                // checkpointing after a chunk left the committed position unpersisted until
                // shutdown whenever the acks landed a moment too late.
                writeCheckpointIfAdvanced();

…red padding Fourth-round review finding, against the current head. The replay cursor advanced only from delivered fragment headers, but poll() advances an image over padding frames without invoking the handler. A recording ending in padding therefore reaches end-of-stream with the cursor still short of the bound, and the outer loop replays the same padding-only range forever, spinning and never rotating. The cursor now advances to what the image actually consumed. This is safe precisely because the read cursor and the commit watermark are separate: it moves only what has been read, while the checkpoint still tracks what has been acknowledged, so nothing unpublished can be committed past. Also adds the first direct test of recording rotation, which until now was implemented and shipped on reasoning alone. What that test does NOT prove, checked rather than assumed: it passes against a build with the cursor fix disabled. Padding in the middle of a recording is stepped over by the next fragment's header and cannot strand the cursor. Only trailing padding can, and producing it needs recording to stop in the window between a padding frame and the message that follows. The livelock therefore remains reasoned, not reproduced, and the test is named for what it actually covers. Tests: 138 in the module, 1 new.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.

Suppressed comments (4)

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:196

A stored checkpoint bypasses the descriptor existence check used for configured recording IDs. If archive retention has removed that recording, both position queries return NULL_POSITION, advanceRecordingIfFinished() treats it as still active, and the source idles forever while appearing healthy. Validate the checkpoint's recording during open() so this fails loudly instead.

            currentRecordingId = stored.get().recordingId();
            currentPosition = stored.get().position();

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:323

This unconditional call defeats checkpointEveryRecords: while caught up, each acknowledgement can advance the watermark and cause a state-store write on the next idle pass, regardless of the configured interval. A failed write is likewise retried and logged on every pass, potentially hammering an unavailable state service and flooding logs. Gate normal writes by the configured committed-record interval, with a separately forced flush for catch-up/rotation/shutdown and backoff for failures.

                // Attempted on every pass, including idle ones. Acknowledgements arrive on
                // framework threads, so the watermark can advance long after the last record was
                // read — and once caught up there is no further chunk to trigger a write. Only
                // checkpointing after a chunk left the committed position unpersisted until
                // shutdown whenever the acks landed a moment too late.
                writeCheckpointIfAdvanced();

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java:52

This documentation still promises a record sequence for broker deduplication, but archive records intentionally expose no sequence (testNoRecordSequenceIsExposed and ArchivePollingRunner document why). Users could enable deduplication expecting effectively-once behavior that is not provided. State the at-least-once behavior and the absence of a deduplication sequence instead.

 *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
 *       storage exists and data missed while the connector was down can be re-read. Records carry
 *       the archive position as their record sequence, which broker deduplication can use.

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java:271

Only -1 is documented as the sentinel for recordingId and startPosition, but other negative values pass validation and are silently treated as unset by resumeOrStart(). A typo such as recordingId: -2 can therefore select the newest recording instead of failing, potentially replaying the wrong range. Reject values below -1 explicitly.

            if (checkpointEveryRecords <= 0) {
                throw new IllegalArgumentException(
                        "checkpointEveryRecords must be positive but was: " + checkpointEveryRecords);

Two findings against 840fd34, both real, both data loss. The FragmentAssembler was created per chunk and discarded at the end of it. On a live recording the chunk bound comes from getRecordingPosition(), which can fall between fragments of a larger-than-MTU message: the assembler held the BEGIN fragment, was thrown away, and the next chunk started on a continuation fragment with a fresh assembler that dropped it. The message vanished silently. This one was created by the previous commit. Before the padding fix the cursor advanced only from delivered fragment headers, so it always sat on a complete-message boundary; advancing to the image position moved it to a fragment boundary and opened the window. The assembler is now held across chunks and reset only when the cursor moves to a different recording, since a partial message cannot span one. Second, concurrent publishers on one channel and stream were silently reduced to one. Transport mode receives from every publication; the archive records each session separately, so concurrent publishers produce concurrent recordings and this replays one at a time. The others would never be replayed, and because discovery only moves to higher recording ids they could not be picked up later either. Archive mode now requires a single active recording and fails at open() with the recording ids and what to do about it, rather than dropping a publisher's data. Historical recordings are unaffected: rotation walks them in id order. Tests: 139 in the module, 1 new. The concurrent-recordings guard was verified by running its test against a build with the guard disabled, where it fails as intended. The assembler fix is NOT directly tested: it needs the chunk bound to land between fragments of a fragmented message on a live recording, which depends on archive write timing and cannot be forced reliably here. Reasoned, not reproduced, and flagged as such.

Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Suppressed comments (6)

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:347

This only verifies that the explicit recording exists. A valid ID belonging to another channel or stream is accepted and replayed, while onFragment() labels every record with the configured channel/stream and successor discovery switches back to that configured stream. Validate the descriptor's stream and channel here, or derive the emitted metadata and rotation scope from the descriptor, to avoid ingesting and mislabeling the wrong recording.

        final int found = archive.listRecording(recordingId,
                (controlSessionId, correlationId, id, startTimestamp, stopTimestamp,
                 startPosition, stopPosition, initialTermId, segmentFileLength, termBufferLength,
                 mtuLength, sessionId, streamId, strippedChannel, originalChannel,
                 sourceIdentity) -> start.set(startPosition));

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java:104

A truncated value is not a confirmed absent checkpoint. Treating corruption as empty restarts from configured history and silently re-ingests data—the same fail-open behavior this class avoids for read exceptions. Fail startup and require an explicit resetCheckpoint instead.

        if (buffer.remaining() < SERIALIZED_BYTES) {
            LOG.warn("Ignoring a truncated archive checkpoint under {} ({} bytes, expected {})",
                    key, buffer.remaining(), SERIALIZED_BYTES);
            return Optional.empty();

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:268

A stored checkpoint bypasses recordingStartPosition(), so its recording ID is never checked for existence. If archive retention has removed that recording, both position queries return NULL_POSITION; the loop then treats it as an active recording and idles forever while the source appears healthy. Validate the checkpoint's recording descriptor (and position range) during open() and fail with reset/recovery guidance when it is no longer replayable.

        if (stored.isPresent()) {
            currentRecordingId = stored.get().recordingId();
            currentPosition = stored.get().position();
            LOG.info("Resuming Aeron archive replay from checkpoint: recordingId={} position={}",
                    currentRecordingId, currentPosition);

aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java:393

This unconditional call bypasses checkpointEveryRecords: any acknowledgement that advances the watermark is persisted on the next loop pass, so a caught-up low-volume stream can perform one state-store write per record even with the default of 1000. Gate normal-loop writes by the configured threshold; rotation and the finally block can still force boundary/final checkpoints.

                writeCheckpointIfAdvanced();

aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java:52

This documents a deduplication capability that the implementation deliberately removed: ArchivePollingRunner never supplies a record sequence, and testNoRecordSequenceIsExposed requires it to be empty because archive positions reset on rotation. As written, users may enable broker deduplication expecting effectively-once delivery even though it cannot operate on these records. Update this contract (and the matching PR description) to describe the position/recording-id properties instead.

 *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
 *       storage exists and data missed while the connector was down can be re-read. Records carry
 *       the archive position as their record sequence, which broker deduplication can use.

aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java:79

The contiguous-prefix algorithm specifically promises correctness for out-of-order framework acknowledgements, but the integration reader acknowledges records serially in emission order and there is no CommitTracker unit test. Add focused tests for out-of-order acknowledgements, a gap/failure blocking later acknowledgements, reset, and duplicate callbacks so this checkpoint-safety invariant is exercised directly.

    synchronized void acked(long position) {
        acked.add(position);
        while (!emitted.isEmpty() && acked.remove(emitted.peekFirst())) {
            committed = emitted.removeFirst();
            anyCommitted = true;

…ng it CI failed testStartsFreshWhenNoCheckpointExists on "running should have produced a checkpoint". The product is fine; the assertion raced. awaitRecords() returns once records are COLLECTED, but the test reader acknowledges after collecting and the checkpoint is written after that, on the poller thread. Asserting the checkpoint immediately races that chain, which is why it passed locally and failed on slower CI hardware. Four assertions had this shape, two of them checking while the source was still running. All now use a shared awaitCheckpoint() helper, whose javadoc says why asserting directly is wrong so the pattern is not reintroduced. The three remaining direct assertions check for ABSENCE of a checkpoint, which is the safe direction, and each already has a settle period. Verified by running the archive suite four times from clean rather than once, since the failure was timing-dependent.

…workaround Sixth-round review finding, and it was about text I wrote. rejectConcurrentRecordings() told the user to "set 'recordingId' explicitly and run one source per recording". That cannot work. The guard runs before the start position is resolved, so it rejects a pinned configuration too — and even if it did not, pinned runners auto-advance on rotation and would walk into each other's recordings. The hard failure stays; the advice goes. Concurrent publishers on one channel and stream are now documented as unsupported in archive mode, with the two remedies that do work: a distinct channel or streamId per publisher, or mode: transport, which genuinely does receive from every publication. Stated in the three places someone meets it — the exception message, the class javadoc, and the mode FieldDoc, which is what the generated config reference shows. A test asserts the message does not offer the pinned-recordingId workaround again.

david-streamlio · 2026-08-12T22:51:23Z

Design question for reviewers: continuous replay vs `ReplayMerge`

Please read this before the diff. It is the one decision in this PR I would most like overruled if
a maintainer disagrees, and the evidence has shifted against my original choice.

The choice I made, and why

Archive mode replays continuous bounded chunks: repeatedly replay
[checkpoint, current recorded position) and advance. #133 originally proposed
ReplayMerge,
which catches up from a recorded position and then transitions into the live stream. I deviated,
and flagged it at the time, for three reasons that still hold:

It cannot outrun durability. Only what the archive has already recorded is consumed, so the
connector never publishes to Pulsar something not yet durable in the archive.
It works over IPC. ReplayMerge needs a multi-destination subscription and a UDP live
destination, so it cannot serve an aeron:ipc channel at all.
The latency it costs is irrelevant here — Pulsar's write path already spends hundreds of
microseconds on fsync, which dwarfs the difference.

What it actually cost

This PR went through six review rounds producing sixteen findings, all of them real, plus three
CI-only failures. That is a lot for one connector mode, so the provenance is worth being precise
about. Five of the sixteen exist only because replay happens in chunks:

Defect	Why chunking caused it
`FragmentAssembler` discarded per chunk → message lost	A chunk bound can fall between fragments of a larger-than-MTU message
Replay cursor stranded by padding → livelock	`poll()` advances the image over padding without invoking the handler, so a cursor driven by delivered headers never reaches the chunk bound
Rotation racing in-flight records → data loss	Recording switch happens at a chunk boundary while acknowledgements are still outstanding
Checkpoint only written after a chunk → stale checkpoint	Acknowledgements land asynchronously; once caught up there is no further chunk to trigger a write
Chunk bound taken from `getRecordingPosition()` on a live recording	The bound is whatever is recorded at that instant, which is not a message boundary

ReplayMerge has a single continuous subscription. None of those five failure modes exist in
it — there is no chunk boundary to land mid-message, no per-chunk assembler, no bound to compute,
and no gap between chunks for a rotation to race into.

So the honest summary of the tradeoff is: this PR accepted five correctness defects in
hand-rolled replay in order to keep IPC support and avoid reading ahead of durability. All five are
fixed, but they were found by review rather than by design, and two of them were introduced by the
fix for a previous one.

What I am asking

Whether that trade is the right one is a maintainer's call, not mine. Three ways it could go:

Keep continuous replay. The three original reasons are real, and IPC support is not
hypothetical — the tests and the demo harness both use aeron:ipc. The defects are fixed. If
this is the answer, I would want the replay loop's invariants written down and property-tested
(see below), because they currently live only in my head and in review comments.
Switch to ReplayMerge. Removes the whole class of chunk-boundary bugs by delegating to code
Aeron maintains and tests. Costs IPC support in archive mode, means the connector can read ahead
of what is durably recorded, and is a substantial rewrite of ArchivePollingRunner — with its
own new review surface.
Both, chosen by channel. ReplayMerge for UDP, chunked replay for IPC. Best behaviour,
worst maintenance: two replay paths, and the chunked one keeps every failure mode above while
getting less real-world exercise.

My own view, having written option 1 and lived through its review: I would not choose it again
without a strong IPC requirement. I underweighted how much hand-rolled cursor management there
was, and how many ways a chunk boundary can be wrong. But I am close to it and may be
over-correcting from a rough review cycle, which is precisely why it wants another opinion.

Regardless of which way it goes

Three paths remain fixed-but-not-reproduced, because I could not construct them reliably against
an in-process archive:

Catalog paging with recording-ID gaps spanning more than one page.
Recordings with a non-zero start position, in both the initial and rotation paths.
Trailing padding stranding the replay cursor — I verified my rotation test still passes with
that fix disabled, so it does not cover it.

And the tests have a systematic blind spot worth fixing whichever design survives: the harness
acknowledges records synchronously, immediately and successfully, which is exactly the shape
that hid the acknowledgement-timing bugs. All three CI-only failures on this PR were the same
mistake appearing in the harness rather than the code. A property-style harness that can produce
delayed, out-of-order and failing acknowledgements would have caught most of this class. Written up
in more detail on #133.

Also worth a reviewer's judgement

Archive mode does not support concurrent publishers on one channel and stream — the archive
records each session separately and this replays one recording at a time, so it fails at startup
when more than one is active. Transport mode has no such limit. That asymmetry is documented and
deliberate, but it is a real semantic gap between the two modes and reasonable people might want it
solved rather than documented.

david-streamlio added 2 commits August 12, 2026 10:33

david-streamlio requested a balanced review from Copilot August 12, 2026 18:00

Copilot started reviewing on behalf of david-streamlio August 12, 2026 18:00 View session