From 26e35c0ddf3ea1d920d615c99029697af7eba23c Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:33:44 -0700 Subject: [PATCH 01/10] [feat][io] Aeron source: Archive-backed lossless mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 #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 #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. --- aeron/build.gradle.kts | 2 + .../apache/pulsar/io/aeron/AeronRecord.java | 19 + .../apache/pulsar/io/aeron/AeronSource.java | 59 +- .../pulsar/io/aeron/AeronSourceConfig.java | 174 ++++++ .../pulsar/io/aeron/ArchiveCheckpoint.java | 120 ++++ .../pulsar/io/aeron/ArchivePollingRunner.java | 400 +++++++++++++ .../AeronArchiveSourceIntegrationTest.java | 559 ++++++++++++++++++ .../io/aeron/AeronSinkContainerTest.java | 5 + .../io/aeron/AeronSourceConfigTest.java | 144 +++++ .../io/aeron/AeronSourceContainerTest.java | 5 + gradle/libs.versions.toml | 2 + 11 files changed, 1478 insertions(+), 11 deletions(-) create mode 100644 aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java create mode 100644 aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java create mode 100644 aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java diff --git a/aeron/build.gradle.kts b/aeron/build.gradle.kts index 3e95fd53e5..5c6314c3ac 100644 --- a/aeron/build.gradle.kts +++ b/aeron/build.gradle.kts @@ -31,6 +31,8 @@ dependencies { // the function runtime, so it is bundled in the NAR. implementation(libs.aeron.driver) implementation(libs.aeron.client) + // Archive mode: replay from a recording rather than the live transport. + implementation(libs.aeron.archive) // The container test drives records into a real broker to prove the full path. testImplementation(libs.testcontainers.pulsar) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java index ac89dab901..6908cdfa2a 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java @@ -44,10 +44,13 @@ public class AeronRecord implements Record { public static final String PROP_CHANNEL = "aeron.channel"; public static final String PROP_POSITION = "aeron.position"; public static final String PROP_INGEST_TS = "aeron.ingest-ts"; + /** Only present in archive mode. */ + public static final String PROP_RECORDING_ID = "aeron.recording-id"; private final byte[] value; private final String key; private final Map properties; + private final Long recordSequence; /** * @param value the reassembled payload; must already be a copy owned by this record @@ -55,9 +58,25 @@ public class AeronRecord implements Record { * @param properties Aeron metadata; copied defensively */ public AeronRecord(byte[] value, String key, Map properties) { + this(value, key, properties, null); + } + + /** + * @param recordSequence the archive position, or null when reading the live transport. Plain + * Aeron has no position that survives a restart, so offering one there + * would imply a resumability the transport does not have. + */ + public AeronRecord(byte[] value, String key, Map properties, + Long recordSequence) { this.value = value; this.key = key; this.properties = Collections.unmodifiableMap(new HashMap<>(properties)); + this.recordSequence = recordSequence; + } + + @Override + public Optional getRecordSequence() { + return Optional.ofNullable(recordSequence); } @Override diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java index 34826735ba..6748099cb4 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSource.java @@ -20,6 +20,7 @@ import io.aeron.Aeron; import io.aeron.Subscription; +import io.aeron.archive.client.AeronArchive; import io.aeron.driver.MediaDriver; import io.aeron.driver.ThreadingMode; import java.util.Map; @@ -39,11 +40,21 @@ * only to a caller that polls: a dedicated thread runs the poll loop and pushes into the * source's internal queue, which the function framework drains via {@code read()}. * - *

Delivery semantics: at-most-once. Plain Aeron is a transport with no persistence - * and no resumable position, so nothing can be replayed. Messages are lost across connector - * restarts, and a multicast subscriber that falls behind loses data once the publisher's term - * buffer rotates. Lossless ingestion needs Aeron Archive, which {@link AeronPoller} leaves - * room for but this implementation does not provide. + *

Delivery semantics depend on the configured mode, and they differ materially. + * + *

    + *
  • {@code transport} (default) reads the live Aeron stream and is + * at-most-once. Plain Aeron has no persistence and no resumable position, so nothing + * can be replayed: messages are lost across connector restarts, and a multicast subscriber + * that falls behind loses data once the publisher's term buffer rotates. + *
  • {@code archive} 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. + *
+ * + *

Because the guarantees differ, the mode is an explicit setting rather than a boolean flag, + * and archive settings supplied in transport mode are rejected rather than ignored — a + * half-configured archive setup that quietly runs at-most-once is the worst available outcome. * *

A single instance owns one subscription; parallelism greater than 1 would give every * instance the same full stream rather than partitioning it. @@ -62,6 +73,7 @@ public class AeronSource extends PushSource { private MediaDriver mediaDriver; private Aeron aeron; private Subscription subscription; + private AeronArchive archive; private AeronPoller poller; private Thread pollerThread; @@ -97,18 +109,33 @@ public void open(Map config, SourceContext sourceContext) throws } aeron = Aeron.connect(aeronContext); - subscription = aeron.addSubscription( - aeronSourceConfig.getChannel(), aeronSourceConfig.getStreamId()); - - poller = new AeronPollingRunner(subscription, aeronSourceConfig, this::consume, sourceContext); + if (aeronSourceConfig.isArchiveMode()) { + archive = AeronArchive.connect(new AeronArchive.Context() + .aeron(aeron) + // The source owns the Aeron client; the archive must not close it. + .ownsAeronClient(false) + .controlRequestChannel(aeronSourceConfig.getArchiveControlRequestChannel()) + .controlResponseChannel( + aeronSourceConfig.getArchiveControlResponseChannel())); + poller = new ArchivePollingRunner( + archive, aeronSourceConfig, this::consume, sourceContext); + } else { + subscription = aeron.addSubscription( + aeronSourceConfig.getChannel(), aeronSourceConfig.getStreamId()); + poller = new AeronPollingRunner( + subscription, aeronSourceConfig, this::consume, sourceContext); + } // A dedicated thread, not a shared pool: the loop runs until close() and would // otherwise occupy a pool thread indefinitely. pollerThread = new Thread(poller, threadName(sourceContext)); pollerThread.setDaemon(true); pollerThread.start(); - LOG.info("Aeron source subscribed to channel {} stream {}", - aeronSourceConfig.getChannel(), aeronSourceConfig.getStreamId()); + // Log the mode: it determines the delivery guarantee, so it must be visible in the + // logs rather than only in the config someone has to go and look up. + LOG.info("Aeron source started in '{}' mode on channel {} stream {}", + aeronSourceConfig.getMode(), aeronSourceConfig.getChannel(), + aeronSourceConfig.getStreamId()); } catch (Exception e) { // open() failing part-way would otherwise leak a media driver process and its // directory, since the framework does not call close() on a failed open(). @@ -170,6 +197,16 @@ private void closeQuietly() { } subscription = null; } + if (archive != null) { + try { + // Constructed with ownsAeronClient(false), so this closes only the archive's own + // control session and leaves the Aeron client for the block below. + archive.close(); + } catch (Exception e) { + LOG.warn("Failed to close Aeron archive client", e); + } + archive = null; + } if (aeron != null) { try { aeron.close(); diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java index 7fb69c9d03..3c2c66a9e7 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java @@ -24,6 +24,10 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Locale; import java.util.Map; import lombok.Data; import lombok.experimental.Accessors; @@ -39,6 +43,25 @@ public class AeronSourceConfig implements Serializable { private static final long serialVersionUID = 1L; + /** Read the live Aeron transport. At-most-once; nothing can be replayed. */ + public static final String MODE_TRANSPORT = "transport"; + /** Replay from an Aeron Archive recording. Recoverable across restarts. */ + public static final String MODE_ARCHIVE = "archive"; + + private static final List MODES = Arrays.asList(MODE_TRANSPORT, MODE_ARCHIVE); + + private static final int DEFAULT_CHECKPOINT_EVERY_RECORDS = 1000; + + @FieldDoc( + required = false, + defaultValue = "transport", + help = "Where the source reads from. 'transport' reads the live Aeron stream and is " + + "at-most-once: nothing can be replayed, so messages published while the " + + "connector is down are lost. 'archive' replays from an Aeron Archive " + + "recording and can resume after a restart. The two have materially " + + "different delivery guarantees, so the mode is explicit rather than a flag") + private String mode = MODE_TRANSPORT; + @FieldDoc( required = true, defaultValue = "", @@ -96,6 +119,78 @@ public class AeronSourceConfig implements Serializable { + "reassembling messages larger than the MTU. 0 uses the Aeron default") private int fragmentAssemblyBufferLength = 0; + // ---- archive mode ---------------------------------------------------------------- + // All null / -1 by default so "the user set this" is distinguishable from "unset", which is + // what lets validate() reject archive settings supplied in transport mode instead of + // silently ignoring them. + + @FieldDoc( + required = false, + defaultValue = "", + help = "Archive control request channel, for example " + + "'aeron:udp?endpoint=localhost:8010'. Required when mode is 'archive'") + private String archiveControlRequestChannel; + + @FieldDoc( + required = false, + defaultValue = "", + help = "Archive control response channel, for example " + + "'aeron:udp?endpoint=localhost:0'. Required when mode is 'archive'") + private String archiveControlResponseChannel; + + @FieldDoc( + required = false, + defaultValue = "-1", + help = "The recording to replay. -1 discovers the most recent recording matching " + + "'channel' and 'streamId'. Only used when mode is 'archive'") + private long recordingId = -1L; + + @FieldDoc( + required = false, + defaultValue = "-1", + help = "Position within the recording to start replaying from. -1 starts at the " + + "beginning of the recording. Only used when mode is 'archive'") + private long startPosition = -1L; + + @FieldDoc( + required = false, + defaultValue = "", + help = "Channel the archive replays onto, for example 'aeron:ipc'. Required when " + + "mode is 'archive'") + private String replayChannel; + + @FieldDoc( + required = false, + defaultValue = "-1", + help = "Stream id used for the replay leg. Must differ from 'streamId' when the " + + "replay channel is the same as the subscription channel. Required when " + + "mode is 'archive'") + private int replayStreamId = -1; + + @FieldDoc( + required = false, + defaultValue = "1000", + help = "How many records to emit between checkpoints. This is the at-least-once " + + "window: a crash replays at most this many records on restart. Lower values " + + "shrink the duplicate window at the cost of more state-store writes. Only " + + "used when mode is 'archive'") + private int checkpointEveryRecords = DEFAULT_CHECKPOINT_EVERY_RECORDS; + + @FieldDoc( + required = false, + defaultValue = "false", + help = "Discards the stored checkpoint at startup and begins again from 'recordingId' " + + "and 'startPosition'. This is a ONE-SHOT operational flag for deliberately " + + "reprocessing history: set it, start the connector once, then REMOVE it. " + + "Left in place it discards the checkpoint on every restart, so the connector " + + "re-ingests the whole recording each time it comes back. Only used when mode " + + "is 'archive'") + private boolean resetCheckpoint = false; + + public boolean isArchiveMode() { + return MODE_ARCHIVE.equalsIgnoreCase(StringUtils.trimToEmpty(mode)); + } + public static AeronSourceConfig load(Map map) throws IOException { ObjectMapper mapper = new ObjectMapper(); return mapper.readValue(mapper.writeValueAsString(map), AeronSourceConfig.class); @@ -143,5 +238,84 @@ public void validate() { throw new IllegalArgumentException( "aeronDirectoryName must be set when useEmbeddedMediaDriver is false"); } + if (mode == null || !MODES.contains(mode.toLowerCase(Locale.ROOT).trim())) { + throw new IllegalArgumentException( + "mode must be one of " + MODES + " but was: " + mode); + } + validateArchiveSettings(); + } + + /** + * Archive settings are required together in archive mode, and rejected outright in transport + * mode. + * + *

Rejecting rather than ignoring is deliberate. A half-configured archive setup that + * silently runs as at-most-once transport is the worst outcome available here: it looks like + * the lossless mode and behaves like the lossy one. + */ + private void validateArchiveSettings() { + if (isArchiveMode()) { + requireArchiveField(archiveControlRequestChannel, "archiveControlRequestChannel"); + requireArchiveField(archiveControlResponseChannel, "archiveControlResponseChannel"); + requireArchiveField(replayChannel, "replayChannel"); + if (replayStreamId == -1) { + throw new IllegalArgumentException( + "replayStreamId must be set when mode is '" + MODE_ARCHIVE + "'"); + } + if (replayStreamId == 0) { + throw new IllegalArgumentException("replayStreamId must not be 0"); + } + if (checkpointEveryRecords <= 0) { + throw new IllegalArgumentException( + "checkpointEveryRecords must be positive but was: " + checkpointEveryRecords); + } + // Same channel and same stream id would make the replay collide with the live + // subscription this source is not even using in archive mode. + if (replayStreamId == streamId && replayChannel.equals(channel)) { + throw new IllegalArgumentException( + "replayStreamId must differ from streamId when replayChannel equals channel"); + } + return; + } + + final List unexpected = new ArrayList<>(); + if (StringUtils.isNotBlank(archiveControlRequestChannel)) { + unexpected.add("archiveControlRequestChannel"); + } + if (StringUtils.isNotBlank(archiveControlResponseChannel)) { + unexpected.add("archiveControlResponseChannel"); + } + if (StringUtils.isNotBlank(replayChannel)) { + unexpected.add("replayChannel"); + } + if (replayStreamId != -1) { + unexpected.add("replayStreamId"); + } + 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"); + } + if (!unexpected.isEmpty()) { + throw new IllegalArgumentException( + "these settings only apply when mode is '" + MODE_ARCHIVE + "': " + unexpected + + ". Set mode to '" + MODE_ARCHIVE + "' or remove them — leaving them " + + "in place with mode '" + MODE_TRANSPORT + "' would run at-most-once " + + "while looking configured for lossless replay"); + } + } + + private static void requireArchiveField(String value, String name) { + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException( + name + " must be set when mode is '" + MODE_ARCHIVE + "'"); + } } } diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java new file mode 100644 index 0000000000..205d5e5e6e --- /dev/null +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.aeron; + +import java.nio.ByteBuffer; +import java.util.Optional; +import org.apache.pulsar.io.core.SourceContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Stores how far the archive replay has progressed, so a restart resumes instead of starting over. + * + *

Persisted through the source's state store, which requires {@code stateStorageServiceUrl} to + * be configured on the cluster. Without it the guarantee cannot be met, so + * {@link #requireAvailable()} fails at {@code open()} rather than letting the connector run and + * silently re-ingest from the beginning on every restart. + * + *

The checkpoint is a recording id and a position, not a bare position. Archive positions + * are only monotonic within one recording: when a publisher restarts, the archive opens a new + * recording numbered from zero, and a position alone would then be meaningless. + */ +final class ArchiveCheckpoint { + + private static final Logger LOG = LoggerFactory.getLogger(ArchiveCheckpoint.class); + + /** recordingId (long) + position (long). */ + private static final int SERIALIZED_BYTES = Long.BYTES * 2; + + private final SourceContext sourceContext; + private final String key; + + ArchiveCheckpoint(SourceContext sourceContext, String channel, int streamId) { + this.sourceContext = sourceContext; + // Keyed by channel and stream so two sources sharing a state namespace do not overwrite + // each other's progress. + this.key = "aeron-archive-position:" + channel + ":" + streamId; + } + + /** A recorded position: which recording, and how far into it. */ + record Position(long recordingId, long position) { } + + /** + * Fails unless the state store is actually usable. + * + *

Checked eagerly because the failure mode otherwise is silent and expensive: the connector + * would appear healthy while replaying the whole recording after every restart. + */ + void requireAvailable() { + try { + sourceContext.getState(key); + } catch (Exception e) { + throw new IllegalStateException( + "Archive mode needs the function state store, but it is unavailable. " + + "Configure 'stateStorageServiceUrl' on the cluster, or the connector " + + "cannot checkpoint and would re-ingest the whole recording on every " + + "restart", e); + } + } + + Optional read() { + try { + final ByteBuffer buffer = sourceContext.getState(key); + if (buffer == null || buffer.remaining() < SERIALIZED_BYTES) { + return Optional.empty(); + } + final ByteBuffer readable = buffer.duplicate(); + return Optional.of(new Position(readable.getLong(), readable.getLong())); + } catch (Exception e) { + // A missing key can surface as an exception rather than null depending on the state + // implementation; treat it as "no checkpoint yet" rather than failing the source. + LOG.debug("No usable archive checkpoint under {}", key, e); + return Optional.empty(); + } + } + + /** + * Discards the stored checkpoint. + * + *

Deleted rather than merely ignored, so the state reflects what actually happened: if the + * operator removes {@code resetCheckpoint} after one run, the next restart resumes normally + * from wherever the reset run reached. + */ + void clear() { + try { + sourceContext.deleteState(key); + } catch (Exception e) { + // Nothing to delete is the common case and not an error. + LOG.debug("Could not delete archive checkpoint under {}", key, e); + } + } + + void write(long recordingId, long position) { + final ByteBuffer buffer = ByteBuffer.allocate(SERIALIZED_BYTES); + buffer.putLong(recordingId); + buffer.putLong(position); + buffer.flip(); + sourceContext.putState(key, buffer); + } + + String key() { + return key; + } +} diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java new file mode 100644 index 0000000000..80d2128871 --- /dev/null +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -0,0 +1,400 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.aeron; + +import io.aeron.Aeron; +import io.aeron.FragmentAssembler; +import io.aeron.Subscription; +import io.aeron.archive.client.AeronArchive; +import io.aeron.logbuffer.Header; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import org.agrona.DirectBuffer; +import org.agrona.collections.MutableLong; +import org.agrona.concurrent.IdleStrategy; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SourceContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Replays an Aeron Archive recording instead of reading the live transport, checkpointing progress + * so a restart resumes where it left off. + * + *

This is the recoverable counterpart to {@link AeronPollingRunner}. Where that reads a live + * stream and can never replay what it missed, this reads durable storage and remembers its place. + * + *

Continuous replay rather than ReplayMerge

+ * + *

The loop repeatedly replays {@code [checkpoint, current recorded position)} and advances the + * checkpoint, rather than using {@link io.aeron.archive.client.ReplayMerge} to catch up and then + * hand over to the live stream. Three reasons: + * + *

    + *
  • It cannot outrun durability. Only what the archive has already recorded is consumed, + * so the connector never publishes to Pulsar something that is not yet durable in the + * archive. For a bridge whose purpose is durable capture, that is the safer behaviour. + *
  • It works over IPC. {@code ReplayMerge} needs a multi-destination subscription and a + * UDP live destination, so it cannot serve an {@code aeron:ipc} channel at all. + *
  • The latency it costs does not matter here. {@code ReplayMerge} wins on steady-state + * latency, but the Pulsar write path already spends hundreds of microseconds on fsync. + *
+ * + *

Where replay starts

+ * + *

In precedence order: + * + *

    + *
  1. {@code resetCheckpoint: true} — discard the stored checkpoint and use {@code recordingId} + * and {@code startPosition}. A one-shot operational flag for deliberately + * reprocessing history: set it, start once, then remove it. Left in place it fires on every + * restart and re-ingests the whole recording each time. + *
  2. A stored checkpoint, if one exists. + *
  3. {@code recordingId} and {@code startPosition}, or discovery of the newest matching + * recording. + *
+ * + *

The checkpoint deliberately outranks a configured {@code startPosition}. The other order + * would silently re-ingest on every restart for anyone who left a {@code startPosition} in their + * config, which is the failure this mode exists to prevent — hence a separate, explicit flag for + * the rare case where starting over is actually what is wanted. + * + *

Delivery semantics: at-least-once

+ * + *

The checkpoint is written after records are handed downstream, so a crash in between replays + * that window on restart. Duplicates are therefore possible and are not deduplicated here. + * Effectively-once is available on top by enabling broker deduplication on the destination topic — + * every record carries its archive position as {@link Record#getRecordSequence()} — but that + * depends on user-side configuration, so it is not promised. + */ +public class ArchivePollingRunner implements AeronPoller { + + private static final Logger LOG = LoggerFactory.getLogger(ArchivePollingRunner.class); + + static final String METRIC_FRAGMENTS_RECEIVED = "aeron-archive-fragments-received"; + static final String METRIC_RECORDS_CONSUMED = "aeron-archive-records-consumed"; + static final String METRIC_CHECKPOINTS_WRITTEN = "aeron-archive-checkpoints-written"; + static final String METRIC_RECORDINGS_ADVANCED = "aeron-archive-recordings-advanced"; + + /** Page size for walking the archive catalog. */ + private static final int LIST_PAGE_SIZE = 100; + + /** Bound on waiting for a replay image, so a stuck replay retries instead of hanging. */ + private static final long REPLAY_CONNECT_TIMEOUT_SECONDS = 30L; + + private final AeronArchive archive; + private final AeronSourceConfig config; + private final Consumer> consumer; + private final SourceContext sourceContext; + private final ArchiveCheckpoint checkpoint; + private final IdleStrategy idleStrategy; + + /** Mutable replay cursor, advanced by the fragment handler. */ + private long currentRecordingId; + private long currentPosition; + private long recordsSinceCheckpoint; + + private volatile boolean running = true; + + public ArchivePollingRunner(AeronArchive archive, + AeronSourceConfig config, + Consumer> consumer, + SourceContext sourceContext) { + this(archive, config, consumer, sourceContext, + new ArchiveCheckpoint(sourceContext, config.getChannel(), config.getStreamId())); + } + + ArchivePollingRunner(AeronArchive archive, + AeronSourceConfig config, + Consumer> consumer, + SourceContext sourceContext, + ArchiveCheckpoint checkpoint) { + this.archive = archive; + this.config = config; + this.consumer = consumer; + this.sourceContext = sourceContext; + this.checkpoint = checkpoint; + this.idleStrategy = IdleStrategies.create(config.getIdleStrategy()); + + checkpoint.requireAvailable(); + resumeOrStart(); + } + + /** + * Picks the starting point: a stored checkpoint wins, then explicit config, then discovery. + * + *

The checkpoint takes precedence deliberately. Honouring a configured {@code startPosition} + * over a stored one would silently re-ingest on every restart, which is the failure this whole + * mode exists to prevent. + */ + private void resumeOrStart() { + if (config.isResetCheckpoint()) { + // Loud on purpose. Left in the config this fires on every restart, and the connector + // re-ingests the whole recording each time it comes back — which looks like a + // duplicate-message problem long before anyone suspects the flag. + LOG.warn("resetCheckpoint is set: discarding any stored checkpoint and restarting from " + + "recordingId={} startPosition={}. This is a one-shot operational " + + "flag — REMOVE it from the config after this run, or every restart " + + "will re-ingest the recording from the beginning.", + config.getRecordingId(), config.getStartPosition()); + checkpoint.clear(); + } + + final Optional stored = + config.isResetCheckpoint() ? Optional.empty() : checkpoint.read(); + if (stored.isPresent()) { + currentRecordingId = stored.get().recordingId(); + currentPosition = stored.get().position(); + LOG.info("Resuming Aeron archive replay from checkpoint: recordingId={} position={}", + currentRecordingId, currentPosition); + return; + } + + currentRecordingId = config.getRecordingId() >= 0 + ? config.getRecordingId() + : discoverRecording(Aeron.NULL_VALUE); + currentPosition = config.getStartPosition() >= 0 ? config.getStartPosition() : 0L; + LOG.info("No checkpoint found; starting Aeron archive replay at recordingId={} position={}", + currentRecordingId, currentPosition); + } + + /** + * Finds a recording for the configured channel and stream. + * + *

Deliberately not {@code findLastMatchingRecording}: it matches its + * {@code sessionId} argument literally rather than treating {@link Aeron#NULL_VALUE} as a + * wildcard, so it only works when the session id is already known — which a connector + * configured with a channel and a stream id never is. {@code listRecordingsForUri} takes no + * session id and filters server-side. + * + * @param after return the lowest recording id strictly greater than this, or the highest + * overall when {@link Aeron#NULL_VALUE} + */ + private long discoverRecording(long after) { + final MutableLong best = new MutableLong(Aeron.NULL_VALUE); + long from = 0; + int matched; + do { + matched = archive.listRecordingsForUri(from, LIST_PAGE_SIZE, + config.getChannel(), config.getStreamId(), + (controlSessionId, correlationId, recordingId, startTimestamp, stopTimestamp, + startPosition, stopPosition, initialTermId, segmentFileLength, + termBufferLength, mtuLength, sessionId, streamId, strippedChannel, + originalChannel, sourceIdentity) -> { + if (after == Aeron.NULL_VALUE) { + if (recordingId > best.get()) { + best.set(recordingId); + } + } else if (recordingId > after + && (best.get() == Aeron.NULL_VALUE || recordingId < best.get())) { + best.set(recordingId); + } + }); + from += matched; + } while (matched == LIST_PAGE_SIZE); + + if (best.get() == Aeron.NULL_VALUE && after == Aeron.NULL_VALUE) { + throw new IllegalStateException( + "No Aeron Archive recording found for channel '" + config.getChannel() + + "' streamId " + config.getStreamId() + + ". Set recordingId explicitly, or check the archive is recording it"); + } + return best.get(); + } + + /** + * How far the given recording can currently be replayed. + * + *

A recording still being written has no stop position — the archive reports + * {@link AeronArchive#NULL_POSITION} — so fall back to how far it has been recorded. Without + * this, a live recording would replay nothing at all, which is the common case rather than an + * edge one. + */ + private long replayableBound(long recordingId) { + final long stop = archive.getStopPosition(recordingId); + return stop != AeronArchive.NULL_POSITION ? stop : archive.getRecordingPosition(recordingId); + } + + @Override + public void run() { + LOG.info("Aeron archive replay loop started for channel {} stream {}", + config.getChannel(), config.getStreamId()); + try { + while (running) { + final long bound = replayableBound(currentRecordingId); + + if (bound > currentPosition) { + replayChunk(bound); + writeCheckpoint(); + } else if (!advanceRecordingIfFinished()) { + // Caught up on a recording that is still being written: wait for more. + idleStrategy.idle(0); + } + } + } catch (Throwable t) { + if (running) { + LOG.error("Aeron archive replay terminated unexpectedly at recordingId={} " + + "position={}", currentRecordingId, currentPosition, t); + } + } finally { + // Best effort: a checkpoint here shrinks the replay window on a clean restart. + try { + writeCheckpoint(); + } catch (Exception e) { + LOG.warn("Could not write a final archive checkpoint", e); + } + LOG.info("Aeron archive replay loop stopped at recordingId={} position={}", + currentRecordingId, currentPosition); + } + } + + /** Replays from the current position up to {@code bound}, advancing the cursor as it goes. */ + private void replayChunk(long bound) { + final long length = bound - currentPosition; + try (Subscription replay = archive.replay( + currentRecordingId, currentPosition, length, + config.getReplayChannel(), config.getReplayStreamId())) { + + final FragmentAssembler assembler = config.getFragmentAssemblyBufferLength() > 0 + ? new FragmentAssembler(this::onFragment, + config.getFragmentAssemblyBufferLength()) + : new FragmentAssembler(this::onFragment); + + // "No image" means two opposite things — not connected yet, and finished then gone + // away — and they are only distinguishable by whether an image was ever seen. Getting + // this wrong ends the chunk before it starts. + boolean sawImage = false; + final long connectDeadline = + System.nanoTime() + TimeUnit.SECONDS.toNanos(REPLAY_CONNECT_TIMEOUT_SECONDS); + + while (running) { + if (replay.imageCount() > 0) { + sawImage = true; + if (replay.imageAtIndex(0).isEndOfStream()) { + break; + } + } else if (sawImage) { + break; // the replay image finished and was removed + } else if (System.nanoTime() > connectDeadline) { + LOG.warn("Replay of recording {} from position {} did not connect within {}s; " + + "retrying", currentRecordingId, currentPosition, + REPLAY_CONNECT_TIMEOUT_SECONDS); + break; + } + + final int fragments = replay.poll(assembler, config.getFragmentLimit()); + if (fragments > 0) { + recordMetric(METRIC_FRAGMENTS_RECEIVED, fragments); + } + idleStrategy.idle(fragments); + } + } + } + + /** + * Moves to the next recording once the current one is finished and fully consumed. + * + *

A publisher restart opens a new recording numbered from zero, which is routine rather than + * exceptional, so the source follows rather than stopping. + * + * @return true if it moved on, false if there is nothing further yet + */ + private boolean advanceRecordingIfFinished() { + final long stop = archive.getStopPosition(currentRecordingId); + if (stop == AeronArchive.NULL_POSITION || currentPosition < stop) { + return false; // still being written, or not caught up + } + + final long next = discoverRecording(currentRecordingId); + if (next == Aeron.NULL_VALUE) { + return false; // this recording is done and no successor exists yet + } + + LOG.info("Recording {} complete at {}; advancing to recording {}", + currentRecordingId, stop, next); + currentRecordingId = next; + // Positions restart per recording, so the cursor must too. + currentPosition = 0L; + recordsSinceCheckpoint = 0; + writeCheckpoint(); + recordMetric(METRIC_RECORDINGS_ADVANCED, 1); + return true; + } + + /** + * Handles one reassembled message. As in the transport runner, the payload must be copied out + * of the buffer before it leaves this callback — Aeron reuses the storage once this returns. + */ + void onFragment(DirectBuffer buffer, int offset, int length, Header header) { + final byte[] payload = new byte[length]; + buffer.getBytes(offset, payload); + + final Map properties = new HashMap<>(); + properties.put(AeronRecord.PROP_SESSION_ID, Integer.toString(header.sessionId())); + properties.put(AeronRecord.PROP_STREAM_ID, Integer.toString(config.getStreamId())); + properties.put(AeronRecord.PROP_CHANNEL, config.getChannel()); + properties.put(AeronRecord.PROP_POSITION, Long.toString(header.position())); + properties.put(AeronRecord.PROP_INGEST_TS, Long.toString(System.currentTimeMillis())); + properties.put(AeronRecord.PROP_RECORDING_ID, Long.toString(currentRecordingId)); + + final String key = config.isKeyBySessionId() ? Integer.toString(header.sessionId()) : null; + + // Positions are monotonic within a recording, which is what broker deduplication needs. + consumer.accept(new AeronRecord(payload, key, properties, header.position())); + recordMetric(METRIC_RECORDS_CONSUMED, 1); + + // The cursor advances only after the record is downstream, so a crash re-replays it rather + // than skipping it. That is the at-least-once side of the trade. + currentPosition = header.position(); + if (++recordsSinceCheckpoint >= config.getCheckpointEveryRecords()) { + writeCheckpoint(); + } + } + + private void writeCheckpoint() { + if (recordsSinceCheckpoint == 0 && currentPosition == 0) { + return; + } + try { + checkpoint.write(currentRecordingId, currentPosition); + recordsSinceCheckpoint = 0; + recordMetric(METRIC_CHECKPOINTS_WRITTEN, 1); + } catch (Exception e) { + // Losing a checkpoint costs a replay, not data, so keep going rather than stopping the + // source — but say so loudly, because a persistently failing store means every restart + // re-ingests from the last good position. + LOG.warn("Failed to write Aeron archive checkpoint at recordingId={} position={}", + currentRecordingId, currentPosition, e); + } + } + + private void recordMetric(String name, int count) { + if (sourceContext != null) { + sourceContext.recordMetric(name, count); + } + } + + @Override + public void stop() { + running = false; + } +} diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java new file mode 100644 index 0000000000..dd826f6dca --- /dev/null +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -0,0 +1,559 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.aeron; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import io.aeron.Aeron; +import io.aeron.Publication; +import io.aeron.archive.Archive; +import io.aeron.archive.ArchivingMediaDriver; +import io.aeron.archive.client.AeronArchive; +import io.aeron.archive.codecs.SourceLocation; +import io.aeron.driver.MediaDriver; +import io.aeron.driver.ThreadingMode; +import java.net.ServerSocket; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import java.util.stream.Collectors; +import org.agrona.concurrent.UnsafeBuffer; +import org.apache.commons.lang3.RandomStringUtils; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SourceContext; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * Exercises {@link AeronSource} in {@code archive} mode against a real Aeron Archive. + * + *

An {@link ArchivingMediaDriver} hosts a media driver and an archive in-process, so the whole + * record-then-replay cycle runs without external services. As with the transport tests, the Aeron + * leg cannot be containerised — a client reaches its driver through memory-mapped files. + * + *

What this pins down is the property the transport mode cannot offer: messages published + * before the source existed are still delivered, because they were recorded. + */ +public class AeronArchiveSourceIntegrationTest { + + private static final String CHANNEL = "aeron:ipc"; + private static final int STREAM_ID = 1001; + private static final int REPLAY_STREAM_ID = 1002; + private static final String REPLAY_CHANNEL = "aeron:ipc"; + private static final long TIMEOUT_SECONDS = 60L; + + private Path rootDir; + private ArchivingMediaDriver archivingMediaDriver; + private Aeron aeron; + private AeronArchive archive; + private AeronSource source; + private Thread readerThread; + private List> collected; + private SourceContext sourceContext; + /** Stands in for the function state store; a bare mock would swallow checkpoints. */ + private Map state; + private String controlRequestChannel; + private String controlResponseChannel; + + @BeforeMethod + public void setUp() throws Exception { + Path parent = Path.of(System.getProperty("aeron.test.dir", + System.getProperty("java.io.tmpdir"))); + Files.createDirectories(parent); + rootDir = Files.createTempDirectory(parent, "aeron-archive-it-"); + collected = new CopyOnWriteArrayList<>(); + state = new ConcurrentHashMap<>(); + sourceContext = mock(SourceContext.class); + // Real read/write semantics, so checkpointing and resume are genuinely exercised rather + // than silently no-oping through a mock. + doAnswer(inv -> { + state.put(inv.getArgument(0), ((ByteBuffer) inv.getArgument(1)).duplicate()); + return null; + }).when(sourceContext).putState(anyString(), any()); + when(sourceContext.getState(anyString())).thenAnswer(inv -> { + ByteBuffer stored = state.get(inv.getArgument(0)); + return stored == null ? null : stored.duplicate(); + }); + + // Ports are picked from the ephemeral range so parallel test JVMs do not collide on a + // fixed archive control port. + final int controlPort = freePort(); + controlRequestChannel = "aeron:udp?endpoint=localhost:" + controlPort; + controlResponseChannel = "aeron:udp?endpoint=localhost:" + freePort(); + + archivingMediaDriver = ArchivingMediaDriver.launch( + new MediaDriver.Context() + .aeronDirectoryName(rootDir.resolve("driver").toString()) + .threadingMode(ThreadingMode.SHARED) + .dirDeleteOnStart(true) + .dirDeleteOnShutdown(true), + new Archive.Context() + .aeronDirectoryName(rootDir.resolve("driver").toString()) + .archiveDir(rootDir.resolve("archive").toFile()) + .controlChannel(controlRequestChannel) + .replicationChannel("aeron:udp?endpoint=localhost:" + freePort()) + .threadingMode(io.aeron.archive.ArchiveThreadingMode.SHARED) + .deleteArchiveOnStart(true)); + + aeron = Aeron.connect(new Aeron.Context() + .aeronDirectoryName(archivingMediaDriver.mediaDriver().aeronDirectoryName())); + archive = AeronArchive.connect(new AeronArchive.Context() + .aeron(aeron) + .ownsAeronClient(false) + .controlRequestChannel(controlRequestChannel) + .controlResponseChannel(controlResponseChannel)); + } + + @AfterMethod(alwaysRun = true) + public void tearDown() { + if (readerThread != null) { + readerThread.interrupt(); + try { + readerThread.join(TimeUnit.SECONDS.toMillis(5)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + readerThread = null; + } + closeQuietly(source); + source = null; + closeQuietly(archive); + archive = null; + closeQuietly(aeron); + aeron = null; + closeQuietly(archivingMediaDriver); + archivingMediaDriver = null; + deleteRecursively(rootDir); + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + private static void closeQuietly(AutoCloseable closeable) { + if (closeable == null) { + return; + } + try { + closeable.close(); + } catch (Exception e) { + // Ignored during teardown. + } + } + + private static void deleteRecursively(Path path) { + if (path == null || !Files.exists(path)) { + return; + } + try (var walk = Files.walk(path)) { + walk.sorted((a, b) -> b.getNameCount() - a.getNameCount()).forEach(p -> { + try { + Files.deleteIfExists(p); + } catch (Exception e) { + // Best effort. + } + }); + } catch (Exception e) { + // Best effort. + } + } + + /** Records a publication and writes the given payloads into it, then stops recording. */ + private void recordMessages(List payloads) { + archive.startRecording(CHANNEL, STREAM_ID, SourceLocation.LOCAL); + try (Publication publication = aeron.addPublication(CHANNEL, STREAM_ID)) { + Awaitility.await("publication connected to the archive recorder") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(50, TimeUnit.MILLISECONDS) + .until(publication::isConnected); + + for (String payload : payloads) { + offer(publication, payload.getBytes(StandardCharsets.UTF_8)); + } + + // Wait for the archive to durably catch up with the publication before replaying; + // otherwise the recording's stop position may still be behind what was published. + final long recordingId = Awaitility.await("recording registered") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(50, TimeUnit.MILLISECONDS) + .until(this::findRecordingId, id -> id != Aeron.NULL_VALUE); + + final long target = publication.position(); + Awaitility.await("archive caught up to " + target) + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(50, TimeUnit.MILLISECONDS) + .until(() -> archive.getRecordingPosition(recordingId) >= target); + } finally { + archive.stopRecording(CHANNEL, STREAM_ID); + } + } + + /** + * Finds the recording by channel and stream. + * + *

Not {@code findLastMatchingRecording}: it matches its sessionId argument literally rather + * than treating {@link Aeron#NULL_VALUE} as a wildcard, so it returns nothing unless the + * session id is already known. Same trap the connector has to avoid. + */ + private long findRecordingId() { + final org.agrona.collections.MutableLong latest = + new org.agrona.collections.MutableLong(Aeron.NULL_VALUE); + archive.listRecordingsForUri(0, 100, CHANNEL, STREAM_ID, + (a, b, recordingId, c, d, e, f, g, h, i, j, k, l, m, n, o) -> { + if (recordingId > latest.get()) { + latest.set(recordingId); + } + }); + return latest.get(); + } + + private static void offer(Publication publication, byte[] payload) { + final UnsafeBuffer buffer = new UnsafeBuffer(payload); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); + while (System.nanoTime() < deadline) { + final long result = publication.offer(buffer, 0, payload.length); + if (result > 0) { + return; + } + if (result == Publication.CLOSED || result == Publication.MAX_POSITION_EXCEEDED) { + fail("Aeron publication unusable: " + result); + } + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + fail("Timed out offering to the recorded publication"); + } + + private Map archiveConfig() { + Map config = new HashMap<>(); + config.put("mode", "archive"); + config.put("channel", CHANNEL); + config.put("streamId", STREAM_ID); + config.put("useEmbeddedMediaDriver", false); + config.put("aeronDirectoryName", archivingMediaDriver.mediaDriver().aeronDirectoryName()); + config.put("idleStrategy", "sleeping"); + config.put("archiveControlRequestChannel", controlRequestChannel); + config.put("archiveControlResponseChannel", + "aeron:udp?endpoint=localhost:0"); + config.put("replayChannel", REPLAY_CHANNEL); + config.put("replayStreamId", REPLAY_STREAM_ID); + return config; + } + + private void startSource(Map config) throws Exception { + source = new AeronSource(); + source.open(config, sourceContext); + readerThread = new Thread(() -> { + try { + while (!Thread.currentThread().isInterrupted()) { + collected.add(source.read()); + } + } catch (Exception e) { + // Interrupted at teardown. + } + }, "aeron-archive-it-reader"); + readerThread.setDaemon(true); + readerThread.start(); + } + + private void awaitRecords(int count) { + Awaitility.await("source emitted " + count + " records") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .until(() -> collected.size() >= count); + } + + private static List valuesOf(List> records) { + return records.stream() + .map(r -> new String(r.getValue(), StandardCharsets.UTF_8)) + .collect(Collectors.toList()); + } + + @Test + public void testReplaysMessagesPublishedBeforeTheSourceExisted() throws Exception { + // The whole point of archive mode: these are written and recorded while no source is + // running at all. Transport mode would deliver none of them. + List sent = new ArrayList<>(); + for (int i = 0; i < 200; i++) { + sent.add("{\"seq\":" + i + ",\"symbol\":\"DEMO\"}"); + } + recordMessages(sent); + + startSource(archiveConfig()); + awaitRecords(sent.size()); + + assertThat(valuesOf(collected)).containsExactlyElementsOf(sent); + } + + @Test + public void testRecordsCarryTheArchivePositionAsRecordSequence() throws Exception { + recordMessages(List.of("alpha", "beta", "gamma")); + + startSource(archiveConfig()); + awaitRecords(3); + + // Monotonic positions are what broker deduplication needs to discard replays. + List sequences = collected.stream() + .map(r -> r.getRecordSequence().orElse(null)) + .collect(Collectors.toList()); + assertThat(sequences).doesNotContainNull(); + assertThat(sequences).isSorted(); + assertThat(sequences.get(sequences.size() - 1)).isGreaterThan(sequences.get(0)); + } + + @Test + public void testRecordsCarryTheRecordingId() throws Exception { + recordMessages(List.of("one")); + + startSource(archiveConfig()); + awaitRecords(1); + + assertThat(collected.get(0).getProperties()) + .containsKey(AeronRecord.PROP_RECORDING_ID) + .containsEntry(AeronRecord.PROP_CHANNEL, CHANNEL) + .containsEntry(AeronRecord.PROP_STREAM_ID, Integer.toString(STREAM_ID)); + assertThat(Long.parseLong( + collected.get(0).getProperties().get(AeronRecord.PROP_RECORDING_ID))) + .isGreaterThanOrEqualTo(0L); + } + + @Test + public void testLargerThanMtuMessagesAreReassembledOnReplay() throws Exception { + String large = RandomStringUtils.insecure().nextAlphanumeric(64 * 1024); + recordMessages(List.of("small", large, "also-small")); + + startSource(archiveConfig()); + awaitRecords(3); + + assertThat(valuesOf(collected)).containsExactly("small", large, "also-small"); + } + + @Test + public void testStartPositionAppliesOnlyWhenThereIsNoCheckpoint() throws Exception { + recordMessages(List.of("first", "second", "third")); + + // Replay everything once to learn where the first message ended. + startSource(archiveConfig()); + awaitRecords(3); + final long afterFirst = collected.get(0).getRecordSequence().orElseThrow(); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + collected.clear(); + + // A stored checkpoint outranks a configured startPosition, so clear it to model a genuinely + // fresh deployment. Without this the source would resume from the checkpoint and deliver + // nothing, which is the correct behaviour and the reason the precedence is that way round: + // honouring startPosition over a checkpoint would re-ingest on every restart. + state.clear(); + + // Positions on a record are end-of-message, so starting from the first record's position + // begins with the message after it. + Map config = archiveConfig(); + config.put("startPosition", afterFirst); + startSource(config); + awaitRecords(2); + + assertThat(valuesOf(collected)).containsExactly("second", "third"); + } + + @Test + public void testCheckpointOutranksConfiguredStartPosition() throws Exception { + // The precedence, asserted directly rather than left implicit: an operator who leaves a + // startPosition in the config must not cause a re-ingest on every restart. + recordMessages(List.of("alpha", "beta")); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(2); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + collected.clear(); + + // Restart with startPosition pointing back at the beginning; the checkpoint should win. + config.put("startPosition", 0L); + startSource(config); + + // Nothing to replay, so nothing should arrive. + Thread.sleep(3000); + assertThat(collected).isEmpty(); + } + + @Test + public void testExplicitRecordingIdIsUsed() throws Exception { + recordMessages(List.of("x", "y")); + final long recordingId = findRecordingId(); + + Map config = archiveConfig(); + config.put("recordingId", recordingId); + startSource(config); + awaitRecords(2); + + assertThat(valuesOf(collected)).containsExactly("x", "y"); + } + + @Test + public void testResumesFromCheckpointAfterRestart() throws Exception { + // The property this whole mode exists for: a restart continues rather than re-ingesting. + List first = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + first.add("first-" + i); + } + recordMessages(first); + + Map config = archiveConfig(); + // Checkpoint aggressively so the restart has something recent to resume from. + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(first.size()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + assertThat(state).as("a checkpoint should have been written").isNotEmpty(); + collected.clear(); + + // More data arrives while the source is down — exactly the window transport mode loses. + List second = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + second.add("second-" + i); + } + recordMessages(second); + + startSource(config); + awaitRecords(second.size()); + + // Only the new messages: the first batch was checkpointed past. + assertThat(valuesOf(collected)).containsExactlyElementsOf(second); + } + + @Test + public void testStartsFreshWhenNoCheckpointExists() throws Exception { + recordMessages(List.of("a", "b", "c")); + + startSource(archiveConfig()); + awaitRecords(3); + + assertThat(valuesOf(collected)).containsExactly("a", "b", "c"); + assertThat(state).as("running should have produced a checkpoint").isNotEmpty(); + } + + @Test + public void testCheckpointSurvivesAsRecordingIdAndPosition() throws Exception { + recordMessages(List.of("one", "two")); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(2); + + // recordingId + position, not a bare position: positions restart per recording, so a + // position alone would be meaningless after a publisher restart. + ByteBuffer stored = state.values().iterator().next().duplicate(); + assertThat(stored.remaining()).isEqualTo(Long.BYTES * 2); + long recordingId = stored.getLong(); + long position = stored.getLong(); + assertThat(recordingId).isGreaterThanOrEqualTo(0L); + assertThat(position).isPositive(); + } + + @Test + public void testResetCheckpointReprocessesFromTheBeginning() throws Exception { + List sent = List.of("alpha", "beta", "gamma"); + recordMessages(sent); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(sent.size()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + assertThat(state).as("first run should have checkpointed").isNotEmpty(); + collected.clear(); + + // The escape hatch: reprocess history despite a checkpoint being present. + config.put("resetCheckpoint", true); + startSource(config); + awaitRecords(sent.size()); + + assertThat(valuesOf(collected)).containsExactlyElementsOf(sent); + } + + @Test + public void testResetCheckpointDiscardsRatherThanIgnoresTheStoredPosition() throws Exception { + recordMessages(List.of("one", "two")); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + config.put("resetCheckpoint", true); + startSource(config); + awaitRecords(2); + + // Deleted, then rewritten as the run progresses — so removing the flag leaves a usable + // checkpoint behind rather than a stale one from before the reset. + assertThat(state).isNotEmpty(); + ByteBuffer stored = state.values().iterator().next().duplicate(); + stored.getLong(); + assertThat(stored.getLong()).as("checkpoint rewritten by the reset run").isPositive(); + } + + @Test + public void testMissingRecordingFailsFast() { + // Nothing has been recorded for this stream, so discovery must fail loudly rather than + // sit silently delivering nothing. + Map config = archiveConfig(); + config.put("streamId", 999); + + AeronSource bad = new AeronSource(); + try { + bad.open(config, sourceContext); + fail("Expected open() to fail when no recording exists"); + } catch (Exception e) { + assertThat(e).isInstanceOf(IllegalStateException.class); + assertThat(e).hasMessageContaining("No Aeron Archive recording found"); + } + } +} diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java index 742b7d2548..a8b099e62b 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java @@ -95,6 +95,11 @@ public class AeronSinkContainerTest { @BeforeMethod public void setUp() throws Exception { pulsarContainer = new PulsarContainer(DockerImageName.parse(PULSAR_IMAGE)) + // Starts the functions worker AND the stream storage (BookKeeper table) service. + // The container's default command appends "--no-functions-worker -nss", and it is + // the -nss ("no stream storage") that would leave the state store absent — which + // the source's archive mode requires for checkpointing. + .withFunctionsWorker() .withStartupTimeout(Duration.ofMinutes(5)); pulsarContainer.start(); diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceConfigTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceConfigTest.java index 6fca035aea..676c1bb8bd 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceConfigTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceConfigTest.java @@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.HashMap; import java.util.Map; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; /** @@ -178,6 +179,149 @@ public void testExternalDriverRequiresDirectory() throws Exception { .hasMessageContaining("aeronDirectoryName must be set"); } + private static Map archiveMap() { + Map map = validMap(); + map.put("mode", "archive"); + map.put("archiveControlRequestChannel", "aeron:udp?endpoint=localhost:8010"); + map.put("archiveControlResponseChannel", "aeron:udp?endpoint=localhost:0"); + map.put("replayChannel", "aeron:ipc"); + map.put("replayStreamId", 2001); + return map; + } + + @Test + public void testDefaultModeIsTransport() throws Exception { + AeronSourceConfig config = AeronSourceConfig.load(validMap()); + + assertThat(config.getMode()).isEqualTo(AeronSourceConfig.MODE_TRANSPORT); + assertThat(config.isArchiveMode()).isFalse(); + config.validate(); + } + + @Test + public void testArchiveModeLoadsAndValidates() throws Exception { + AeronSourceConfig config = AeronSourceConfig.load(archiveMap()); + config.validate(); + + assertThat(config.isArchiveMode()).isTrue(); + assertThat(config.getReplayStreamId()).isEqualTo(2001); + assertThat(config.getRecordingId()).isEqualTo(-1L); + assertThat(config.getStartPosition()).isEqualTo(-1L); + } + + @Test + public void testArchiveModeIsCaseInsensitive() throws Exception { + Map map = archiveMap(); + map.put("mode", "ARCHIVE"); + + AeronSourceConfig config = AeronSourceConfig.load(map); + config.validate(); + + assertThat(config.isArchiveMode()).isTrue(); + } + + @Test + public void testUnknownModeIsRejected() throws Exception { + Map map = validMap(); + map.put("mode", "replay"); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("mode must be one of"); + } + + @DataProvider(name = "requiredArchiveFields") + public static Object[][] requiredArchiveFields() { + return new Object[][]{ + {"archiveControlRequestChannel"}, + {"archiveControlResponseChannel"}, + {"replayChannel"}, + {"replayStreamId"}, + }; + } + + @Test(dataProvider = "requiredArchiveFields") + public void testArchiveModeRequiresItsSettings(String field) throws Exception { + Map map = archiveMap(); + map.remove(field); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(field); + } + + @DataProvider(name = "archiveOnlyFields") + public static Object[][] archiveOnlyFields() { + return new Object[][]{ + {"archiveControlRequestChannel", "aeron:udp?endpoint=localhost:8010"}, + {"archiveControlResponseChannel", "aeron:udp?endpoint=localhost:0"}, + {"replayChannel", "aeron:ipc"}, + {"replayStreamId", 2001}, + {"recordingId", 7L}, + {"startPosition", 128L}, + {"resetCheckpoint", true}, + {"checkpointEveryRecords", 50}, + }; + } + + @Test(dataProvider = "archiveOnlyFields") + public void testArchiveSettingsAreRejectedInTransportMode(String field, Object value) + throws Exception { + // Rejected rather than ignored: a config that looks set up for lossless replay but runs + // at-most-once is the worst outcome available, so it must fail at open() instead. + Map map = validMap(); + map.put(field, value); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(field) + .hasMessageContaining("only apply when mode is 'archive'"); + } + + @Test + public void testResetCheckpointDefaultsOff() throws Exception { + // Default matters: this flag re-ingests the whole recording every restart if left on. + AeronSourceConfig config = AeronSourceConfig.load(archiveMap()); + config.validate(); + + assertThat(config.isResetCheckpoint()).isFalse(); + assertThat(config.getCheckpointEveryRecords()).isEqualTo(1000); + } + + @Test + public void testNonPositiveCheckpointIntervalIsRejected() throws Exception { + Map map = archiveMap(); + map.put("checkpointEveryRecords", 0); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("checkpointEveryRecords must be positive"); + } + + @Test + public void testZeroReplayStreamIdIsRejected() throws Exception { + Map map = archiveMap(); + map.put("replayStreamId", 0); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("replayStreamId must not be 0"); + } + + @Test + public void testReplayCollidingWithTheSubscriptionIsRejected() throws Exception { + // Same channel and same stream id would make the replay collide with the stream being + // replayed from. + Map map = archiveMap(); + map.put("channel", "aeron:ipc"); + map.put("replayChannel", "aeron:ipc"); + map.put("replayStreamId", map.get("streamId")); + + assertThatThrownBy(() -> AeronSourceConfig.load(map).validate()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("replayStreamId must differ from streamId"); + } + @Test public void testEmbeddedDriverDoesNotRequireDirectory() throws Exception { Map map = validMap(); diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java index bde29dce74..7fe4df45af 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java @@ -94,6 +94,11 @@ public class AeronSourceContainerTest { @BeforeMethod public void setUp() throws Exception { pulsarContainer = new PulsarContainer(DockerImageName.parse(PULSAR_IMAGE)) + // Starts the functions worker AND the stream storage (BookKeeper table) service. + // The container's default command appends "--no-functions-worker -nss", and it is + // the -nss ("no stream storage") that would leave the state store absent — which + // the source's archive mode requires for checkpointing. + .withFunctionsWorker() .withStartupTimeout(Duration.ofMinutes(5)); pulsarContainer.start(); diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 9d162c7774..088190f926 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -229,6 +229,8 @@ netty-reactive-streams = { module = "com.typesafe.netty:netty-reactive-streams", # the source connector does not need. aeron-client = { module = "io.aeron:aeron-client", version.ref = "aeron" } aeron-driver = { module = "io.aeron:aeron-driver", version.ref = "aeron" } +# Aeron Archive: recording and replay, for the source's lossless archive mode. +aeron-archive = { module = "io.aeron:aeron-archive", version.ref = "aeron" } # Protobuf / gRPC protobuf-bom = { module = "com.google.protobuf:protobuf-bom", version.ref = "protobuf" } protobuf-java = { module = "com.google.protobuf:protobuf-java" } From 4dcba00c907d741e317ad4cb2d8852653b656d33 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:59:00 -0700 Subject: [PATCH 02/10] [fix][test] Aeron container tests: wait for the namespace, not just the 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. --- .../pulsar/io/aeron/AeronSinkContainerTest.java | 10 ++++++++++ .../pulsar/io/aeron/AeronSourceContainerTest.java | 12 ++++++++++-- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java index a8b099e62b..d0b4789351 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSinkContainerTest.java @@ -110,11 +110,21 @@ public void setUp() throws Exception { .serviceHttpUrl(pulsarContainer.getHttpServiceUrl()) .build(); + // Bootstrap creates the "public" tenant and the "public/default" namespace + // asynchronously, and the container's wait strategy can return before either lands. + // Waiting on the tenant alone is not enough: the namespace is created after it, so a + // producer can still fail with "Namespace not found". Enabling the functions worker + // shifted the timing enough to make that visible in CI. Awaitility.await("public tenant created") .atMost(60, TimeUnit.SECONDS) .pollInterval(250, TimeUnit.MILLISECONDS) .ignoreExceptions() .until(() -> admin.tenants().getTenants().contains("public")); + Awaitility.await("public/default namespace created") + .atMost(60, TimeUnit.SECONDS) + .pollInterval(250, TimeUnit.MILLISECONDS) + .ignoreExceptions() + .until(() -> admin.namespaces().getNamespaces("public").contains("public/default")); topicName = "persistent://public/default/aggregates-" + System.nanoTime(); producer = pulsarClient.newProducer(Schema.BYTES).topic(topicName).create(); diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java index 7fe4df45af..0d6c63422d 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronSourceContainerTest.java @@ -109,13 +109,21 @@ public void setUp() throws Exception { .serviceHttpUrl(pulsarContainer.getHttpServiceUrl()) .build(); - // The broker creates the "public" tenant asynchronously during bootstrap and the - // container wait strategy can return before that lands. + // Bootstrap creates the "public" tenant and the "public/default" namespace + // asynchronously, and the container's wait strategy can return before either lands. + // Waiting on the tenant alone is not enough: the namespace is created after it, so a + // producer can still fail with "Namespace not found". Enabling the functions worker + // shifted the timing enough to make that visible in CI. Awaitility.await("public tenant created") .atMost(60, TimeUnit.SECONDS) .pollInterval(250, TimeUnit.MILLISECONDS) .ignoreExceptions() .until(() -> admin.tenants().getTenants().contains("public")); + Awaitility.await("public/default namespace created") + .atMost(60, TimeUnit.SECONDS) + .pollInterval(250, TimeUnit.MILLISECONDS) + .ignoreExceptions() + .until(() -> admin.namespaces().getNamespaces("public").contains("public/default")); topicName = "persistent://public/default/aeron-in-" + System.nanoTime(); producer = pulsarClient.newProducer(Schema.BYTES).topic(topicName).create(); From 46fc978f7cd89a8619823983ef61951dbc5dc814 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:25:08 -0700 Subject: [PATCH 03/10] [fix][io] Aeron archive mode: address review findings on #134 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 #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. --- .../apache/pulsar/io/aeron/AeronRecord.java | 52 ++++-- .../pulsar/io/aeron/ArchiveCheckpoint.java | 19 ++- .../pulsar/io/aeron/ArchivePollingRunner.java | 103 ++++++++++-- .../apache/pulsar/io/aeron/CommitTracker.java | 115 +++++++++++++ .../AeronArchiveSourceIntegrationTest.java | 151 ++++++++++++++++-- 5 files changed, 395 insertions(+), 45 deletions(-) create mode 100644 aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java index 6908cdfa2a..6fecae2ca8 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronRecord.java @@ -33,9 +33,14 @@ * callback returns, so the bytes must be copied before the record leaves the polling * thread. See {@link AeronPollingRunner}. * - *

{@link #ack()} and {@link #fail()} are deliberately no-ops. Plain Aeron is a transport - * with no persistence and no resumable position, so there is nothing to acknowledge and - * nothing to redeliver. This is what makes the connector at-most-once. + *

In transport mode {@link #ack()} and {@link #fail()} are no-ops: plain Aeron has no + * persistence and no resumable position, so there is nothing to acknowledge and nothing to + * redeliver, which is what makes that mode at-most-once. + * + *

In archive mode they are not decoration. The framework queues a record via {@code consume()} + * and only publishes it later, so an acknowledgement is the sole evidence that a record reached + * Pulsar. Committing a replay position on anything earlier would checkpoint past records that were + * never published. */ public class AeronRecord implements Record { @@ -47,10 +52,23 @@ public class AeronRecord implements Record { /** Only present in archive mode. */ public static final String PROP_RECORDING_ID = "aeron.recording-id"; + /** + * Told whether the framework managed to publish a record. + * + *

Archive mode needs this: the framework publishes and acknowledges after + * {@code consume()} has queued the record, so only an acknowledgement proves a record is + * safely in Pulsar and its position safe to checkpoint. + */ + interface Outcome { + void acked(); + + void failed(); + } + private final byte[] value; private final String key; private final Map properties; - private final Long recordSequence; + private final Outcome outcome; /** * @param value the reassembled payload; must already be a copy owned by this record @@ -62,21 +80,14 @@ public AeronRecord(byte[] value, String key, Map properties) { } /** - * @param recordSequence the archive position, or null when reading the live transport. Plain - * Aeron has no position that survives a restart, so offering one there - * would imply a resumability the transport does not have. + * @param outcome notified when the framework publishes or fails this record, or null for the + * live transport, where there is no position to commit and nothing to redeliver */ - public AeronRecord(byte[] value, String key, Map properties, - Long recordSequence) { + AeronRecord(byte[] value, String key, Map properties, Outcome outcome) { this.value = value; this.key = key; this.properties = Collections.unmodifiableMap(new HashMap<>(properties)); - this.recordSequence = recordSequence; - } - - @Override - public Optional getRecordSequence() { - return Optional.ofNullable(recordSequence); + this.outcome = outcome; } @Override @@ -96,11 +107,18 @@ public Map getProperties() { @Override public void ack() { - // No-op: see class javadoc. Plain Aeron has nothing to acknowledge. + // Transport mode has nothing to acknowledge; archive mode commits its position here, + // which is the only point at which the record is known to be in Pulsar. + if (outcome != null) { + outcome.acked(); + } } @Override public void fail() { - // No-op: see class javadoc. Plain Aeron cannot redeliver a message. + // Transport mode cannot redeliver. Archive mode must not checkpoint past this record. + if (outcome != null) { + outcome.failed(); + } } } diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java index 205d5e5e6e..a5dbd709b4 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java @@ -98,11 +98,26 @@ Optional read() { * from wherever the reset run reached. */ void clear() { + Exception deleteFailure = null; try { sourceContext.deleteState(key); } catch (Exception e) { - // Nothing to delete is the common case and not an error. - LOG.debug("Could not delete archive checkpoint under {}", key, e); + // Deleting an absent key can legitimately throw depending on the state + // implementation, so the exception alone does not mean the reset failed. Verify by + // reading instead of guessing. + deleteFailure = e; + } + + if (read().isPresent()) { + throw new IllegalStateException( + "resetCheckpoint was requested but the stored checkpoint under '" + key + + "' could not be deleted. Continuing would resume from the old " + + "position while reporting a reset, so the source is failing instead.", + deleteFailure); + } + if (deleteFailure != null) { + LOG.debug("deleteState threw for {} but no checkpoint remains, so the reset succeeded", + key, deleteFailure); } } diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 80d2128871..7ce687f1b4 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -80,11 +80,26 @@ * *

Delivery semantics: at-least-once

* - *

The checkpoint is written after records are handed downstream, so a crash in between replays - * that window on restart. Duplicates are therefore possible and are not deduplicated here. - * Effectively-once is available on top by enabling broker deduplication on the destination topic — - * every record carries its archive position as {@link Record#getRecordSequence()} — but that - * depends on user-side configuration, so it is not promised. + *

Two cursors are tracked, and the distinction is the whole guarantee. The replay cursor + * is how far the loop has read. The commit watermark, maintained by {@link CommitTracker}, + * is the highest position whose record — and every record before it — the framework has + * acknowledged, meaning it actually reached Pulsar. Only the watermark is checkpointed. + * + *

This matters because {@code consume()} merely queues a record; the framework publishes and + * acknowledges later. Checkpointing the replay cursor would commit positions for records still + * sitting in memory, and a crash in that window would resume past records that were never + * published — silent loss in the mode meant to prevent exactly that. + * + *

A crash between publish and checkpoint replays the window since the last checkpoint, so + * duplicates are possible; {@code checkpointEveryRecords} bounds it. A record the framework + * fails stalls the watermark rather than being stepped over, so a restart replays it. + * + *

No record sequence is set. An earlier revision exposed the archive position as + * {@link Record#getRecordSequence()} so broker deduplication could give effectively-once, but + * archive positions restart with each recording. After a rotation the same producer would emit + * decreasing sequence ids, and Pulsar's deduplication would then discard the new + * recording as already-seen — turning an optional optimisation into active message loss. Doing + * this safely needs a sequence that is monotonic across rotation, which is follow-up work. */ public class ArchivePollingRunner implements AeronPoller { @@ -107,6 +122,7 @@ public class ArchivePollingRunner implements AeronPoller { private final SourceContext sourceContext; private final ArchiveCheckpoint checkpoint; private final IdleStrategy idleStrategy; + private CommitTracker commitTracker; /** Mutable replay cursor, advanced by the fragment handler. */ private long currentRecordingId; @@ -137,6 +153,7 @@ public ArchivePollingRunner(AeronArchive archive, checkpoint.requireAvailable(); resumeOrStart(); + this.commitTracker = new CommitTracker(currentPosition); } /** @@ -172,7 +189,10 @@ private void resumeOrStart() { currentRecordingId = config.getRecordingId() >= 0 ? config.getRecordingId() : discoverRecording(Aeron.NULL_VALUE); - currentPosition = config.getStartPosition() >= 0 ? config.getStartPosition() : 0L; + // Default to where the recording actually begins rather than assuming zero. + currentPosition = config.getStartPosition() >= 0 + ? config.getStartPosition() + : recordingStartPosition(currentRecordingId); LOG.info("No checkpoint found; starting Aeron archive replay at recordingId={} position={}", currentRecordingId, currentPosition); } @@ -191,15 +211,18 @@ private void resumeOrStart() { */ private long discoverRecording(long after) { final MutableLong best = new MutableLong(Aeron.NULL_VALUE); + final MutableLong highestSeen = new MutableLong(Aeron.NULL_VALUE); long from = 0; int matched; do { + highestSeen.set(Aeron.NULL_VALUE); matched = archive.listRecordingsForUri(from, LIST_PAGE_SIZE, config.getChannel(), config.getStreamId(), (controlSessionId, correlationId, recordingId, startTimestamp, stopTimestamp, startPosition, stopPosition, initialTermId, segmentFileLength, termBufferLength, mtuLength, sessionId, streamId, strippedChannel, originalChannel, sourceIdentity) -> { + highestSeen.set(Math.max(highestSeen.get(), recordingId)); if (after == Aeron.NULL_VALUE) { if (recordingId > best.get()) { best.set(recordingId); @@ -209,7 +232,12 @@ private long discoverRecording(long after) { best.set(recordingId); } }); - from += matched; + // The first argument is an inclusive recording ID, not an offset. Advancing it by the + // match count re-reads the same page whenever matching IDs have gaps, which loops + // forever; step past the highest ID this page actually returned instead. + if (matched > 0 && highestSeen.get() != Aeron.NULL_VALUE) { + from = highestSeen.get() + 1; + } } while (matched == LIST_PAGE_SIZE); if (best.get() == Aeron.NULL_VALUE && after == Aeron.NULL_VALUE) { @@ -221,6 +249,22 @@ private long discoverRecording(long after) { return best.get(); } + /** + * Where the given recording actually begins. + * + *

Not necessarily zero: Aeron supports recordings with a non-zero start position, and + * asking {@code replay()} to start before one fails asynchronously rather than at the call. + */ + private long recordingStartPosition(long recordingId) { + final MutableLong start = new MutableLong(0L); + archive.listRecording(recordingId, + (controlSessionId, correlationId, id, startTimestamp, stopTimestamp, + startPosition, stopPosition, initialTermId, segmentFileLength, termBufferLength, + mtuLength, sessionId, streamId, strippedChannel, originalChannel, + sourceIdentity) -> start.set(startPosition)); + return start.get(); + } + /** * How far the given recording can currently be replayed. * @@ -254,6 +298,12 @@ public void run() { if (running) { LOG.error("Aeron archive replay terminated unexpectedly at recordingId={} " + "position={}", currentRecordingId, currentPosition, t); + // Without this the thread dies while AeronSource stays open and read() blocks + // forever, so a dead connector reports itself healthy. Telling the runtime lets it + // restart the instance, which resumes from the checkpoint. + if (sourceContext != null) { + sourceContext.fatal(t); + } } } finally { // Best effort: a checkpoint here shrinks the replay window on a clean restart. @@ -332,8 +382,10 @@ private boolean advanceRecordingIfFinished() { LOG.info("Recording {} complete at {}; advancing to recording {}", currentRecordingId, stop, next); currentRecordingId = next; - // Positions restart per recording, so the cursor must too. - currentPosition = 0L; + // Positions restart per recording — but not necessarily at zero, so ask the successor + // where it begins rather than assuming. + currentPosition = recordingStartPosition(next); + commitTracker.reset(currentPosition); recordsSinceCheckpoint = 0; writeCheckpoint(); recordMetric(METRIC_RECORDINGS_ADVANCED, 1); @@ -357,25 +409,42 @@ void onFragment(DirectBuffer buffer, int offset, int length, Header header) { properties.put(AeronRecord.PROP_RECORDING_ID, Long.toString(currentRecordingId)); final String key = config.isKeyBySessionId() ? Integer.toString(header.sessionId()) : null; + final long position = header.position(); + + // Registered before handing the record over: an acknowledgement can land on a framework + // thread the instant consume() returns, and the tracker must already know about it. + commitTracker.emitted(position); + consumer.accept(new AeronRecord(payload, key, properties, new AeronRecord.Outcome() { + @Override + public void acked() { + commitTracker.acked(position); + } - // Positions are monotonic within a recording, which is what broker deduplication needs. - consumer.accept(new AeronRecord(payload, key, properties, header.position())); + @Override + public void failed() { + commitTracker.failed(position); + } + })); recordMetric(METRIC_RECORDS_CONSUMED, 1); - // The cursor advances only after the record is downstream, so a crash re-replays it rather - // than skipping it. That is the at-least-once side of the trade. - currentPosition = header.position(); + // The replay cursor advances here so the loop knows how far it has read. What gets + // *checkpointed* is the commit watermark, which only moves on acknowledgement. + currentPosition = position; if (++recordsSinceCheckpoint >= config.getCheckpointEveryRecords()) { writeCheckpoint(); } } private void writeCheckpoint() { - if (recordsSinceCheckpoint == 0 && currentPosition == 0) { + // The committed watermark, never the replay cursor: positions between the two belong to + // records that are queued but not yet published, and committing those would resume past + // data that never reached Pulsar. + if (!commitTracker.hasCommitted()) { return; } + final long committed = commitTracker.committedPosition(); try { - checkpoint.write(currentRecordingId, currentPosition); + checkpoint.write(currentRecordingId, committed); recordsSinceCheckpoint = 0; recordMetric(METRIC_CHECKPOINTS_WRITTEN, 1); } catch (Exception e) { @@ -383,7 +452,7 @@ private void writeCheckpoint() { // source — but say so loudly, because a persistently failing store means every restart // re-ingests from the last good position. LOG.warn("Failed to write Aeron archive checkpoint at recordingId={} position={}", - currentRecordingId, currentPosition, e); + currentRecordingId, committed, e); } } diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java new file mode 100644 index 0000000000..0290278a41 --- /dev/null +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.aeron; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashSet; +import java.util.Set; + +/** + * Tracks which replayed positions have actually reached Pulsar, so only those are checkpointed. + * + *

Why this exists

+ * + *

{@code PushSource.consume()} merely queues a record. The framework drains that queue, publishes, + * and only then calls {@link org.apache.pulsar.functions.api.Record#ack()}. Checkpointing at + * enqueue time would therefore commit positions for records still sitting in memory, and a + * crash in that window would resume past records that were never published — silent loss, in + * the mode whose entire purpose is not losing anything. + * + *

So the replay cursor and the committed cursor are separate. This class turns + * out-of-order acknowledgements into a contiguous committed watermark: the highest position such + * that every position emitted before it has been acknowledged. + * + *

Failures stall rather than skip

+ * + *

A failed record is left in place. The watermark stops behind it and stops advancing, so the + * next restart replays from before the failure instead of stepping over it. A permanently failing + * record therefore halts checkpoint progress, which is the correct trade for a lossless mode: + * repeated work is recoverable, skipped data is not. + * + *

Acknowledgements arrive on framework threads while emissions happen on the poller thread, so + * every method is synchronized. The critical sections are a few pointer moves and are dwarfed by + * the Pulsar publish they accompany. + */ +final class CommitTracker { + + private final Deque emitted = new ArrayDeque<>(); + private final Set acked = new HashSet<>(); + + private long committed; + private boolean anyCommitted; + private long failures; + + CommitTracker(long startPosition) { + this.committed = startPosition; + } + + /** Records that a position has been handed to the framework but is not yet published. */ + synchronized void emitted(long position) { + emitted.addLast(position); + } + + /** Marks a position published, then advances the watermark over any contiguous run. */ + synchronized void acked(long position) { + acked.add(position); + while (!emitted.isEmpty() && acked.remove(emitted.peekFirst())) { + committed = emitted.removeFirst(); + anyCommitted = true; + } + } + + /** + * Marks a position as failed to publish. + * + *

Deliberately does not remove it from the queue: leaving it there is what stalls the + * watermark and keeps the record replayable. + */ + synchronized void failed(long position) { + failures++; + } + + /** The highest position whose record — and every record before it — reached Pulsar. */ + synchronized long committedPosition() { + return committed; + } + + /** True once anything has been committed, so an untouched start position is not persisted. */ + synchronized boolean hasCommitted() { + return anyCommitted; + } + + /** Records handed downstream but neither acknowledged nor failed yet. */ + synchronized int pending() { + return emitted.size(); + } + + synchronized long failureCount() { + return failures; + } + + /** Drops all tracking, for when the cursor jumps — a recording rotation or a reset. */ + synchronized void reset(long startPosition) { + emitted.clear(); + acked.clear(); + committed = startPosition; + anyCommitted = false; + } +} diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index dd826f6dca..44c0319e41 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -84,6 +84,8 @@ public class AeronArchiveSourceIntegrationTest { private SourceContext sourceContext; /** Stands in for the function state store; a bare mock would swallow checkpoints. */ private Map state; + /** Lets a test withhold acknowledgements to model records stuck mid-publish. */ + private volatile boolean ackRecords = true; private String controlRequestChannel; private String controlResponseChannel; @@ -94,6 +96,7 @@ public void setUp() throws Exception { Files.createDirectories(parent); rootDir = Files.createTempDirectory(parent, "aeron-archive-it-"); collected = new CopyOnWriteArrayList<>(); + ackRecords = true; state = new ConcurrentHashMap<>(); sourceContext = mock(SourceContext.class); // Real read/write semantics, so checkpointing and resume are genuinely exercised rather @@ -106,6 +109,12 @@ public void setUp() throws Exception { ByteBuffer stored = state.get(inv.getArgument(0)); return stored == null ? null : stored.duplicate(); }); + // deleteState was previously left as Mockito's no-op, which made every reset test + // vacuous: they would have passed even if clear() deleted nothing at all. + doAnswer(inv -> { + state.remove(inv.getArgument(0)); + return null; + }).when(sourceContext).deleteState(anyString()); // Ports are picked from the ephemeral range so parallel test JVMs do not collide on a // fixed archive control port. @@ -279,7 +288,16 @@ private void startSource(Map config) throws Exception { readerThread = new Thread(() -> { try { while (!Thread.currentThread().isInterrupted()) { - collected.add(source.read()); + Record record = source.read(); + collected.add(record); + // The framework acknowledges after a successful publish, and only an + // acknowledgement advances the commit watermark. A reader that collected + // without acking modelled a world where publishing always succeeds + // instantly — which is precisely why the earlier revision's checkpointing + // bug was invisible to these tests. + if (ackRecords) { + record.ack(); + } } } catch (Exception e) { // Interrupted at teardown. @@ -319,19 +337,25 @@ public void testReplaysMessagesPublishedBeforeTheSourceExisted() throws Exceptio } @Test - public void testRecordsCarryTheArchivePositionAsRecordSequence() throws Exception { + public void testNoRecordSequenceIsExposed() throws Exception { recordMessages(List.of("alpha", "beta", "gamma")); startSource(archiveConfig()); awaitRecords(3); - // Monotonic positions are what broker deduplication needs to discard replays. - List sequences = collected.stream() - .map(r -> r.getRecordSequence().orElse(null)) + // Deliberately absent. Archive positions restart with each recording, so after a rotation + // the same producer would emit decreasing sequence ids and Pulsar's deduplication would + // discard the new recording as already-seen — turning an optional optimisation into active + // message loss. A sequence monotonic across rotation is follow-up work. + assertThat(collected).allSatisfy(r -> + assertThat(r.getRecordSequence()).isEmpty()); + + // The position is still available as metadata; only the dedup affordance is withheld. + List positions = collected.stream() + .map(r -> Long.parseLong(r.getProperties().get(AeronRecord.PROP_POSITION))) .collect(Collectors.toList()); - assertThat(sequences).doesNotContainNull(); - assertThat(sequences).isSorted(); - assertThat(sequences.get(sequences.size() - 1)).isGreaterThan(sequences.get(0)); + assertThat(positions).isSorted(); + assertThat(positions.get(positions.size() - 1)).isGreaterThan(positions.get(0)); } @Test @@ -368,7 +392,8 @@ public void testStartPositionAppliesOnlyWhenThereIsNoCheckpoint() throws Excepti // Replay everything once to learn where the first message ended. startSource(archiveConfig()); awaitRecords(3); - final long afterFirst = collected.get(0).getRecordSequence().orElseThrow(); + final long afterFirst = + Long.parseLong(collected.get(0).getProperties().get(AeronRecord.PROP_POSITION)); readerThread.interrupt(); readerThread = null; @@ -540,6 +565,114 @@ public void testResetCheckpointDiscardsRatherThanIgnoresTheStoredPosition() thro assertThat(stored.getLong()).as("checkpoint rewritten by the reset run").isPositive(); } + @Test + public void testUnacknowledgedRecordsAreNotCheckpointed() throws Exception { + // The regression test for the defect this design exists to prevent. consume() only queues + // a record; the framework publishes and acks later. If the checkpoint tracked the replay + // cursor instead of the acknowledged watermark, a crash here would resume PAST records + // that never reached Pulsar. + ackRecords = false; + recordMessages(List.of("un-acked-1", "un-acked-2", "un-acked-3")); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(3); + + // Read and delivered, but never acknowledged — so nothing may be committed. + Thread.sleep(2000); + assertThat(state) + .as("no checkpoint may exist while every record is still unacknowledged") + .isEmpty(); + } + + @Test + public void testCheckpointOnlyAdvancesOverTheAcknowledgedPrefix() throws Exception { + recordMessages(List.of("a", "b", "c")); + + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(3); + + // With every record acked, the watermark reaches the last position, so a restart has + // nothing left to replay. + Awaitility.await("checkpoint written") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> !state.isEmpty()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + collected.clear(); + + startSource(config); + Thread.sleep(3000); + assertThat(collected).as("everything was acknowledged, so nothing should replay").isEmpty(); + } + + @Test + public void testResetFailsLoudlyWhenTheCheckpointCannotBeDeleted() throws Exception { + recordMessages(List.of("x")); + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(1); + Awaitility.await("checkpoint written") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> !state.isEmpty()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + + // A state store that accepts the delete but does not actually remove anything: silently + // continuing would resume from the old position while reporting a reset. + doAnswer(inv -> null).when(sourceContext).deleteState(anyString()); + + config.put("resetCheckpoint", true); + AeronSource bad = new AeronSource(); + try { + bad.open(config, sourceContext); + fail("Expected open() to fail when the checkpoint could not be deleted"); + } catch (Exception e) { + assertThat(e).isInstanceOf(IllegalStateException.class); + assertThat(e).hasMessageContaining("could not be deleted"); + } finally { + closeQuietly(bad); + } + } + + @Test + public void testResetActuallyDeletesTheStoredCheckpoint() throws Exception { + // Guards the stub itself: with deleteState left as a Mockito no-op this passed vacuously. + recordMessages(List.of("p", "q")); + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(2); + Awaitility.await("checkpoint written") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> !state.isEmpty()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + collected.clear(); + + // Withhold acks so the reset run cannot immediately write a replacement checkpoint, + // leaving the deletion itself observable. + ackRecords = false; + config.put("resetCheckpoint", true); + startSource(config); + awaitRecords(2); + + assertThat(state).as("the stored checkpoint must actually be gone").isEmpty(); + } + @Test public void testMissingRecordingFailsFast() { // Nothing has been recorded for this stream, so discovery must fail loudly rather than From c0bd4c5c2b7cf83de42f6eeddcd44694fcfd5249 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:40:34 -0700 Subject: [PATCH 04/10] [fix][io] Aeron archive mode: checkpoint on idle passes, not only after a chunk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../pulsar/io/aeron/ArchivePollingRunner.java | 24 +++++++++++++++---- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 7ce687f1b4..1a51937180 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -128,6 +128,8 @@ public class ArchivePollingRunner implements AeronPoller { private long currentRecordingId; private long currentPosition; private long recordsSinceCheckpoint; + private long lastCheckpointedPosition = Long.MIN_VALUE; + private long lastCheckpointedRecording = Aeron.NULL_VALUE; private volatile boolean running = true; @@ -288,11 +290,17 @@ public void run() { if (bound > currentPosition) { replayChunk(bound); - writeCheckpoint(); } else if (!advanceRecordingIfFinished()) { // Caught up on a recording that is still being written: wait for more. idleStrategy.idle(0); } + + // 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(); } } catch (Throwable t) { if (running) { @@ -308,7 +316,7 @@ public void run() { } finally { // Best effort: a checkpoint here shrinks the replay window on a clean restart. try { - writeCheckpoint(); + writeCheckpointIfAdvanced(); } catch (Exception e) { LOG.warn("Could not write a final archive checkpoint", e); } @@ -387,7 +395,7 @@ private boolean advanceRecordingIfFinished() { currentPosition = recordingStartPosition(next); commitTracker.reset(currentPosition); recordsSinceCheckpoint = 0; - writeCheckpoint(); + writeCheckpointIfAdvanced(); recordMetric(METRIC_RECORDINGS_ADVANCED, 1); return true; } @@ -431,11 +439,12 @@ public void failed() { // *checkpointed* is the commit watermark, which only moves on acknowledgement. currentPosition = position; if (++recordsSinceCheckpoint >= config.getCheckpointEveryRecords()) { - writeCheckpoint(); + writeCheckpointIfAdvanced(); } } - private void writeCheckpoint() { + /** Writes only when the watermark has actually moved, so idle passes stay cheap. */ + private void writeCheckpointIfAdvanced() { // The committed watermark, never the replay cursor: positions between the two belong to // records that are queued but not yet published, and committing those would resume past // data that never reached Pulsar. @@ -443,8 +452,13 @@ private void writeCheckpoint() { return; } final long committed = commitTracker.committedPosition(); + if (committed == lastCheckpointedPosition && currentRecordingId == lastCheckpointedRecording) { + return; + } try { checkpoint.write(currentRecordingId, committed); + lastCheckpointedPosition = committed; + lastCheckpointedRecording = currentRecordingId; recordsSinceCheckpoint = 0; recordMetric(METRIC_CHECKPOINTS_WRITTEN, 1); } catch (Exception e) { From 2b0c52ded617e69bd25b94cb56030e7ccb338986 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:14:49 -0700 Subject: [PATCH 05/10] [fix][io] Aeron archive mode: second-round review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../pulsar/io/aeron/ArchiveCheckpoint.java | 54 +++++++++++++--- .../pulsar/io/aeron/ArchivePollingRunner.java | 39 +++++++++++- .../AeronArchiveSourceIntegrationTest.java | 63 +++++++++++++++++++ 3 files changed, 145 insertions(+), 11 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java index a5dbd709b4..08aa9a999e 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java @@ -74,20 +74,37 @@ void requireAvailable() { } } + /** + * Reads the stored checkpoint. + * + *

Returns empty only for a confirmed absent or unreadably short value. A store + * failure propagates rather than being reported as "no checkpoint": treating a transient read + * error as an absent checkpoint would silently restart from configured history and re-ingest + * the recording, which is precisely the failure archive mode exists to prevent. Failing + * startup is recoverable; silently re-ingesting looks like success. + * + * @throws IllegalStateException if the state store could not be read + */ Optional read() { + final ByteBuffer buffer; try { - final ByteBuffer buffer = sourceContext.getState(key); - if (buffer == null || buffer.remaining() < SERIALIZED_BYTES) { - return Optional.empty(); - } - final ByteBuffer readable = buffer.duplicate(); - return Optional.of(new Position(readable.getLong(), readable.getLong())); + buffer = sourceContext.getState(key); } catch (Exception e) { - // A missing key can surface as an exception rather than null depending on the state - // implementation; treat it as "no checkpoint yet" rather than failing the source. - LOG.debug("No usable archive checkpoint under {}", key, e); + throw new IllegalStateException( + "Could not read the Aeron archive checkpoint under '" + key + "'. Refusing to " + + "continue, because treating this as an absent checkpoint would " + + "re-ingest the recording from the configured start position.", e); + } + if (buffer == null) { return Optional.empty(); } + if (buffer.remaining() < SERIALIZED_BYTES) { + LOG.warn("Ignoring a truncated archive checkpoint under {} ({} bytes, expected {})", + key, buffer.remaining(), SERIALIZED_BYTES); + return Optional.empty(); + } + final ByteBuffer readable = buffer.duplicate(); + return Optional.of(new Position(readable.getLong(), readable.getLong())); } /** @@ -108,7 +125,24 @@ void clear() { deleteFailure = e; } - if (read().isPresent()) { + // Verify rather than assume. read() now propagates store failures, so a delete that + // failed AND a verification read that failed can no longer be mistaken for a successful + // reset — the read throws and the reset is reported as failed, which is correct. + final boolean stillPresent; + try { + stillPresent = read().isPresent(); + } catch (Exception readFailure) { + final IllegalStateException error = new IllegalStateException( + "resetCheckpoint was requested but the stored checkpoint under '" + key + + "' could neither be deleted nor verified, so whether the reset took " + + "effect is unknown. Failing rather than guessing.", readFailure); + if (deleteFailure != null) { + error.addSuppressed(deleteFailure); + } + throw error; + } + + if (stillPresent) { throw new IllegalStateException( "resetCheckpoint was requested but the stored checkpoint under '" + key + "' could not be deleted. Continuing would resume from the old " diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 1a51937180..21ec6c120e 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -116,6 +116,10 @@ public class ArchivePollingRunner implements AeronPoller { /** Bound on waiting for a replay image, so a stuck replay retries instead of hanging. */ private static final long REPLAY_CONNECT_TIMEOUT_SECONDS = 30L; + /** Throttle for the rotation-blocked warning, which would otherwise spin. */ + private static final long ROTATION_STALL_LOG_INTERVAL_NANOS = + TimeUnit.SECONDS.toNanos(30); + private final AeronArchive archive; private final AeronSourceConfig config; private final Consumer> consumer; @@ -130,6 +134,7 @@ public class ArchivePollingRunner implements AeronPoller { private long recordsSinceCheckpoint; private long lastCheckpointedPosition = Long.MIN_VALUE; private long lastCheckpointedRecording = Aeron.NULL_VALUE; + private long lastRotationStallLogNanos; private volatile boolean running = true; @@ -259,11 +264,20 @@ private long discoverRecording(long after) { */ private long recordingStartPosition(long recordingId) { final MutableLong start = new MutableLong(0L); - archive.listRecording(recordingId, + 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)); + if (found == 0) { + // Returning a synthetic 0 here would be worse than failing: every subsequent position + // query answers NULL_POSITION, the loop treats the recording as merely still being + // written, and the source idles forever while reporting itself healthy. + throw new IllegalStateException( + "Aeron Archive has no recording with id " + recordingId + + ". Check 'recordingId', or leave it unset to discover the newest " + + "recording for the configured channel and stream"); + } return start.get(); } @@ -382,6 +396,29 @@ private boolean advanceRecordingIfFinished() { return false; // still being written, or not caught up } + // Rotating with records still awaiting acknowledgement would be unsafe twice over. + // Resetting the tracker discards those pending entries, so their positions could never be + // committed; and because positions restart in the successor, a late acknowledgement for an + // old position can collide with a successor position and advance ITS watermark, skipping + // unpublished successor records after a crash. So drain first. + final int inFlight = commitTracker.pending(); + if (inFlight > 0) { + final long now = System.nanoTime(); + if (now - lastRotationStallLogNanos > ROTATION_STALL_LOG_INTERVAL_NANOS) { + lastRotationStallLogNanos = now; + LOG.warn("Recording {} is complete but {} record(s) are still awaiting " + + "acknowledgement, so rotation is waiting. A record the framework " + + "never acknowledges will hold this indefinitely, which is " + + "deliberate: advancing would skip unpublished data.", + currentRecordingId, inFlight); + } + return false; + } + + // Persist the old recording's final watermark before the cursor moves, so a crash during + // the switch resumes at its end rather than somewhere earlier. + writeCheckpointIfAdvanced(); + final long next = discoverRecording(currentRecordingId); if (next == Aeron.NULL_VALUE) { return false; // this recording is done and no successor exists yet diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index 44c0319e41..b41d38d6af 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -673,6 +673,69 @@ public void testResetActuallyDeletesTheStoredCheckpoint() throws Exception { assertThat(state).as("the stored checkpoint must actually be gone").isEmpty(); } + @Test + public void testNonexistentRecordingIdFailsFastRatherThanIdling() throws Exception { + // Without the descriptor check this returned a synthetic position 0, every subsequent + // position query answered NULL_POSITION, and the loop idled forever while the source + // reported itself healthy and emitted nothing. + recordMessages(List.of("present")); + + Map config = archiveConfig(); + config.put("recordingId", 9999L); + + AeronSource bad = new AeronSource(); + try { + bad.open(config, sourceContext); + fail("Expected open() to reject a recordingId that does not exist"); + } catch (Exception e) { + assertThat(e).isInstanceOf(IllegalStateException.class); + assertThat(e).hasMessageContaining("no recording with id 9999"); + } finally { + closeQuietly(bad); + } + } + + @Test + public void testUnreadableCheckpointStoreFailsRatherThanReingesting() throws Exception { + // A transient read failure previously read as "no checkpoint", which would restart from + // configured history and silently re-ingest the recording. + recordMessages(List.of("m1", "m2")); + Map config = archiveConfig(); + config.put("checkpointEveryRecords", 1); + startSource(config); + awaitRecords(2); + Awaitility.await("checkpoint written") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> !state.isEmpty()); + + readerThread.interrupt(); + readerThread = null; + source.close(); + source = null; + + // requireAvailable() succeeds, then the read for the checkpoint itself blows up. + final java.util.concurrent.atomic.AtomicInteger reads = + new java.util.concurrent.atomic.AtomicInteger(); + when(sourceContext.getState(anyString())).thenAnswer(inv -> { + if (reads.incrementAndGet() > 1) { + throw new IllegalStateException("state store unavailable"); + } + ByteBuffer stored = state.get(inv.getArgument(0)); + return stored == null ? null : stored.duplicate(); + }); + + AeronSource bad = new AeronSource(); + try { + bad.open(config, sourceContext); + fail("Expected open() to fail when the checkpoint could not be read"); + } catch (Exception e) { + assertThat(e).isInstanceOf(IllegalStateException.class); + assertThat(e).hasMessageContaining("Could not read the Aeron archive checkpoint"); + } finally { + closeQuietly(bad); + } + } + @Test public void testMissingRecordingFailsFast() { // Nothing has been recorded for this stream, so discovery must fail loudly rather than From 276a2523200717e428ac316c1af57b784270a3b4 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:46:45 -0700 Subject: [PATCH 06/10] [fix][io] Aeron archive mode: fail the source on a publish failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../pulsar/io/aeron/ArchivePollingRunner.java | 23 +++++++++++++-- .../apache/pulsar/io/aeron/CommitTracker.java | 10 +++++-- .../AeronArchiveSourceIntegrationTest.java | 29 ++++++++++++++++++- 3 files changed, 56 insertions(+), 6 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 21ec6c120e..26c92165dc 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -91,8 +91,13 @@ * published — silent loss in the mode meant to prevent exactly that. * *

A crash between publish and checkpoint replays the window since the last checkpoint, so - * duplicates are possible; {@code checkpointEveryRecords} bounds it. A record the framework - * fails stalls the watermark rather than being stepped over, so a restart replays it. + * duplicates are possible; {@code checkpointEveryRecords} bounds it. + * + *

A record the framework fails stalls the watermark and fails the source. + * Stalling alone would not be enough: the replay cursor has already moved past the record, so + * nothing would re-emit it and the message would stay absent from Pulsar while the connector + * looked healthy. Failing the instance lets the runtime restart it and replay from the + * checkpoint. * *

No record sequence is set. An earlier revision exposed the archive position as * {@link Record#getRecordSequence()} so broker deduplication could give effectively-once, but @@ -468,6 +473,20 @@ public void acked() { @Override public void failed() { commitTracker.failed(position); + // Stalling the watermark stops the failure being checkpointed past, but nothing + // re-emits the record in this process: the replay cursor has moved on, so the + // message would stay absent from Pulsar while the source looked healthy. That is + // not at-least-once. Failing the instance lets the runtime restart it, which + // resumes from the checkpoint and replays the record. Same approach as the + // Kinesis source. + LOG.error("Publishing a record at recordingId={} position={} failed; failing the " + + "source so it restarts and replays from the last checkpoint", + currentRecordingId, position); + if (sourceContext != null) { + sourceContext.fatal(new IllegalStateException( + "Failed to publish Aeron archive record at recordingId=" + + currentRecordingId + " position=" + position)); + } } })); recordMetric(METRIC_RECORDS_CONSUMED, 1); diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java index 0290278a41..136f6a73c2 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java @@ -41,9 +41,13 @@ *

Failures stall rather than skip

* *

A failed record is left in place. The watermark stops behind it and stops advancing, so the - * next restart replays from before the failure instead of stepping over it. A permanently failing - * record therefore halts checkpoint progress, which is the correct trade for a lossless mode: - * repeated work is recoverable, skipped data is not. + * next restart replays from before the failure instead of stepping over it. + * + *

This is only half the answer, and on its own it would not deliver at-least-once: the replay + * cursor has already moved past the record, so nothing re-emits it within this process. The caller + * therefore also fails the source on a publish failure, so the runtime restarts it and the + * stalled watermark decides where replay resumes. This class holds the position; it does not + * decide the policy. * *

Acknowledgements arrive on framework threads while emissions happen on the poller thread, so * every method is synchronized. The critical sections are a few pointer moves and are dwarfed by diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index b41d38d6af..c2702512a4 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -22,7 +22,9 @@ import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import io.aeron.Aeron; @@ -86,6 +88,8 @@ public class AeronArchiveSourceIntegrationTest { private Map state; /** Lets a test withhold acknowledgements to model records stuck mid-publish. */ private volatile boolean ackRecords = true; + /** Models a downstream publish failure. */ + private volatile boolean failRecords; private String controlRequestChannel; private String controlResponseChannel; @@ -97,6 +101,7 @@ public void setUp() throws Exception { rootDir = Files.createTempDirectory(parent, "aeron-archive-it-"); collected = new CopyOnWriteArrayList<>(); ackRecords = true; + failRecords = false; state = new ConcurrentHashMap<>(); sourceContext = mock(SourceContext.class); // Real read/write semantics, so checkpointing and resume are genuinely exercised rather @@ -295,7 +300,9 @@ private void startSource(Map config) throws Exception { // without acking modelled a world where publishing always succeeds // instantly — which is precisely why the earlier revision's checkpointing // bug was invisible to these tests. - if (ackRecords) { + if (failRecords) { + record.fail(); + } else if (ackRecords) { record.ack(); } } @@ -736,6 +743,26 @@ public void testUnreadableCheckpointStoreFailsRatherThanReingesting() throws Exc } } + @Test + public void testPublishFailureFailsTheSourceSoItRestartsAndReplays() throws Exception { + // Stalling the watermark alone is not at-least-once: the replay cursor has already moved + // past the record, so nothing re-emits it and the message would stay absent from Pulsar + // while the source looked healthy. The instance must fail so the runtime restarts it. + failRecords = true; + recordMessages(List.of("will-fail")); + + startSource(archiveConfig()); + awaitRecords(1); + + Awaitility.await("source reported the failure as fatal") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> verify(sourceContext, atLeastOnce()).fatal(any())); + + // And nothing was committed, so a restart replays the failed record. + assertThat(state).as("a failed record must not be checkpointed").isEmpty(); + } + @Test public void testMissingRecordingFailsFast() { // Nothing has been recorded for this stream, so discovery must fail loudly rather than From 840fd3400873975d362205214a24bc82bac8fe93 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:38:29 -0700 Subject: [PATCH 07/10] [fix][io] Aeron archive mode: advance the replay cursor over undelivered 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. --- .../pulsar/io/aeron/ArchivePollingRunner.java | 18 ++++++- .../AeronArchiveSourceIntegrationTest.java | 53 ++++++++++++++++++- 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 26c92165dc..0290f831b9 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -20,6 +20,7 @@ import io.aeron.Aeron; import io.aeron.FragmentAssembler; +import io.aeron.Image; import io.aeron.Subscription; import io.aeron.archive.client.AeronArchive; import io.aeron.logbuffer.Header; @@ -360,13 +361,18 @@ private void replayChunk(long bound) { // away — and they are only distinguishable by whether an image was ever seen. Getting // this wrong ends the chunk before it starts. boolean sawImage = false; + // Tracks what the replay image actually consumed, which is not the same as what was + // delivered: poll() advances over padding frames without calling the handler. + long imagePosition = currentPosition; final long connectDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(REPLAY_CONNECT_TIMEOUT_SECONDS); while (running) { if (replay.imageCount() > 0) { sawImage = true; - if (replay.imageAtIndex(0).isEndOfStream()) { + final Image image = replay.imageAtIndex(0); + imagePosition = Math.max(imagePosition, image.position()); + if (image.isEndOfStream()) { break; } } else if (sawImage) { @@ -384,6 +390,16 @@ private void replayChunk(long bound) { } idleStrategy.idle(fragments); } + + // Advance the replay cursor over anything the image consumed but did not deliver. + // A range ending in padding delivers no fragment, so a cursor driven only by fragment + // headers would stay short of the bound and the outer loop would replay the same + // padding-only range forever, never rotating. Safe precisely because the cursor and + // the commit watermark are separate: this moves only what has been *read*, while the + // checkpoint still tracks what has been acknowledged. + if (imagePosition > currentPosition) { + currentPosition = imagePosition; + } } } diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index c2702512a4..7203674836 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -208,8 +208,17 @@ private static void deleteRecursively(Path path) { /** Records a publication and writes the given payloads into it, then stops recording. */ private void recordMessages(List payloads) { + recordMessages(payloads, CHANNEL); + } + + /** + * @param publicationChannel lets a test pick a small term length, so terms roll over and Aeron + * inserts padding frames — which {@code poll()} skips without + * invoking the fragment handler + */ + private void recordMessages(List payloads, String publicationChannel) { archive.startRecording(CHANNEL, STREAM_ID, SourceLocation.LOCAL); - try (Publication publication = aeron.addPublication(CHANNEL, STREAM_ID)) { + try (Publication publication = aeron.addPublication(publicationChannel, STREAM_ID)) { Awaitility.await("publication connected to the archive recorder") .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) .pollInterval(50, TimeUnit.MILLISECONDS) @@ -763,6 +772,48 @@ public void testPublishFailureFailsTheSourceSoItRestartsAndReplays() throws Exce assertThat(state).as("a failed record must not be checkpointed").isEmpty(); } + @Test + public void testAdvancesToTheNextRecordingWhenTheCurrentOneIsFinished() throws Exception { + // Rotation-following had no direct test at all until this one; it was implemented and + // shipped on reasoning alone. + // + // The small term length makes terms roll over, so padding frames occur mid-recording and + // the replay has to cross them. Note what this does NOT prove: padding in the middle of a + // recording is stepped over by the next fragment's header, so it cannot strand the replay + // cursor. Only padding at the very END of a recording can, and that needs recording to + // stop in the window between a padding frame being written and the message that follows + // it. This test was checked against a build with the cursor fix disabled and still passed, + // so the trailing-padding livelock remains reasoned rather than reproduced. + final String paddedChannel = CHANNEL + "?term-length=65536"; + List first = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + first.add("first-" + i + "-" + RandomStringUtils.insecure().nextAlphanumeric(6000)); + } + recordMessages(first, paddedChannel); + // Discovery picks the NEWEST recording, so the start point has to be pinned to the first + // one — otherwise the source begins at the second and rotation is never exercised. + final long firstRecordingId = findRecordingId(); + + // A second recording, which the source can only reach if it got past the first. + List second = List.of("second-a", "second-b"); + recordMessages(second, paddedChannel); + assertThat(findRecordingId()) + .as("the second batch must create a distinct recording") + .isGreaterThan(firstRecordingId); + + Map config = archiveConfig(); + config.put("recordingId", firstRecordingId); + startSource(config); + awaitRecords(first.size() + second.size()); + + List values = valuesOf(collected); + assertThat(values).hasSize(first.size() + second.size()); + assertThat(values.subList(0, first.size())).containsExactlyElementsOf(first); + assertThat(values.subList(first.size(), values.size())) + .as("the source must reach the second recording rather than spinning on the first") + .containsExactlyElementsOf(second); + } + @Test public void testMissingRecordingFailsFast() { // Nothing has been recorded for this stream, so discovery must fail loudly rather than From 6455e28947370d1cf8a7b7c7c8066e7382e20c0c Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:07:53 -0700 Subject: [PATCH 08/10] [fix][io] Aeron archive mode: fifth-round review findings 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. --- .../pulsar/io/aeron/ArchivePollingRunner.java | 86 +++++++++++++++++-- .../AeronArchiveSourceIntegrationTest.java | 49 +++++++++++ 2 files changed, 130 insertions(+), 5 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 0290f831b9..95c6fa836f 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -100,6 +100,14 @@ * looked healthy. Failing the instance lets the runtime restart it and replay from the * checkpoint. * + *

One active recording at a time

+ * + *

Transport mode receives from every publication on a channel and stream. Archive mode cannot: + * the archive records each session separately, so concurrent publishers produce concurrent + * recordings and this replays one at a time. Rather than silently dropping the others, it fails at + * {@code open()} when more than one is active. Give each publisher a distinct channel or stream, + * or pin {@code recordingId} and run a source per recording. + * *

No record sequence is set. An earlier revision exposed the archive position as * {@link Record#getRecordSequence()} so broker deduplication could give effectively-once, but * archive positions restart with each recording. After a rotation the same producer would emit @@ -133,6 +141,16 @@ public class ArchivePollingRunner implements AeronPoller { private final ArchiveCheckpoint checkpoint; private final IdleStrategy idleStrategy; private CommitTracker commitTracker; + /** + * Reassembly state, deliberately held across chunks rather than created per chunk. + * + *

On a live recording the chunk bound comes from {@code getRecordingPosition()}, which can + * fall between fragments of a larger-than-MTU message. A per-chunk assembler would + * hold the BEGIN fragment, be discarded at the end of the chunk, and the next chunk would + * start on a continuation fragment with a fresh assembler that drops it — losing the message + * silently. Reset only when the cursor jumps to a different recording. + */ + private FragmentAssembler assembler; /** Mutable replay cursor, advanced by the fragment handler. */ private long currentRecordingId; @@ -165,8 +183,60 @@ public ArchivePollingRunner(AeronArchive archive, this.idleStrategy = IdleStrategies.create(config.getIdleStrategy()); checkpoint.requireAvailable(); + rejectConcurrentRecordings(); resumeOrStart(); this.commitTracker = new CommitTracker(currentPosition); + this.assembler = newAssembler(); + } + + /** + * Fails when more than one recording for this channel and stream is currently active. + * + *

Transport mode subscribes to a channel and stream and receives from every + * publication on it. Archive mode cannot match that: 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 ever moves to + * higher recording ids, lower ones could never be picked up later either. + * + *

Rather than silently dropping those publishers' data, archive mode requires a single + * active recording and says so. Historical recordings are fine: rotation walks them in id + * order, though it replays each in full rather than interleaving them as the live transport + * would. + */ + private void rejectConcurrentRecordings() { + final MutableLong active = new MutableLong(0); + final StringBuilder ids = new StringBuilder(); + long from = 0; + int matched; + final MutableLong highestSeen = new MutableLong(Aeron.NULL_VALUE); + do { + highestSeen.set(Aeron.NULL_VALUE); + matched = archive.listRecordingsForUri(from, LIST_PAGE_SIZE, + config.getChannel(), config.getStreamId(), + (controlSessionId, correlationId, recordingId, startTimestamp, stopTimestamp, + startPosition, stopPosition, initialTermId, segmentFileLength, + termBufferLength, mtuLength, sessionId, streamId, strippedChannel, + originalChannel, sourceIdentity) -> { + highestSeen.set(Math.max(highestSeen.get(), recordingId)); + if (stopPosition == AeronArchive.NULL_POSITION) { + active.set(active.get() + 1); + ids.append(ids.length() == 0 ? "" : ", ").append(recordingId); + } + }); + if (matched > 0 && highestSeen.get() != Aeron.NULL_VALUE) { + from = highestSeen.get() + 1; + } + } while (matched == LIST_PAGE_SIZE); + + if (active.get() > 1) { + throw new IllegalStateException( + "Archive mode found " + active.get() + " concurrently active recordings for " + + "channel '" + config.getChannel() + "' streamId " + + config.getStreamId() + " (recordingIds: " + ids + "). It replays one " + + "recording at a time, so the others would never be replayed. Use a " + + "distinct channel or streamId per publisher, or set 'recordingId' " + + "explicitly and run one source per recording."); + } } /** @@ -345,6 +415,15 @@ public void run() { } } + /** + * Builds the reassembly buffer. Held across chunks rather than per chunk — see the field. + */ + private FragmentAssembler newAssembler() { + return config.getFragmentAssemblyBufferLength() > 0 + ? new FragmentAssembler(this::onFragment, config.getFragmentAssemblyBufferLength()) + : new FragmentAssembler(this::onFragment); + } + /** Replays from the current position up to {@code bound}, advancing the cursor as it goes. */ private void replayChunk(long bound) { final long length = bound - currentPosition; @@ -352,11 +431,6 @@ private void replayChunk(long bound) { currentRecordingId, currentPosition, length, config.getReplayChannel(), config.getReplayStreamId())) { - final FragmentAssembler assembler = config.getFragmentAssemblyBufferLength() > 0 - ? new FragmentAssembler(this::onFragment, - config.getFragmentAssemblyBufferLength()) - : new FragmentAssembler(this::onFragment); - // "No image" means two opposite things — not connected yet, and finished then gone // away — and they are only distinguishable by whether an image was ever seen. Getting // this wrong ends the chunk before it starts. @@ -452,6 +526,8 @@ private boolean advanceRecordingIfFinished() { // where it begins rather than assuming. currentPosition = recordingStartPosition(next); commitTracker.reset(currentPosition); + // A partially assembled message cannot continue into a different recording. + assembler = newAssembler(); recordsSinceCheckpoint = 0; writeCheckpointIfAdvanced(); recordMetric(METRIC_RECORDINGS_ADVANCED, 1); diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index 7203674836..4d5432a479 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -814,6 +814,55 @@ public void testAdvancesToTheNextRecordingWhenTheCurrentOneIsFinished() throws E .containsExactlyElementsOf(second); } + @Test + public void testConcurrentActiveRecordingsAreRejected() throws Exception { + // Two publishers on one channel and stream produce two recordings, because the archive + // records each session separately. Archive mode replays one at a time, so silently + // choosing one would drop the other publisher's data entirely. + archive.startRecording(CHANNEL, STREAM_ID, SourceLocation.LOCAL); + // Exclusive publications, because two ordinary ones on the same channel and stream share + // a single publication. Each exclusive publication gets its own session, which is what + // makes the archive create a separate recording per publisher. + try (Publication first = aeron.addExclusivePublication(CHANNEL, STREAM_ID); + Publication second = aeron.addExclusivePublication(CHANNEL, STREAM_ID)) { + Awaitility.await("both publications connected") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> first.isConnected() && second.isConnected()); + offer(first, "from-first".getBytes(StandardCharsets.UTF_8)); + offer(second, "from-second".getBytes(StandardCharsets.UTF_8)); + + Awaitility.await("two active recordings registered") + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .until(this::countActiveRecordings, count -> count >= 2); + + AeronSource bad = new AeronSource(); + try { + bad.open(archiveConfig(), sourceContext); + fail("Expected open() to reject concurrently active recordings"); + } catch (Exception e) { + assertThat(e).isInstanceOf(IllegalStateException.class); + assertThat(e).hasMessageContaining("concurrently active recordings"); + } finally { + closeQuietly(bad); + } + } finally { + archive.stopRecording(CHANNEL, STREAM_ID); + } + } + + private int countActiveRecordings() { + final org.agrona.collections.MutableInteger active = + new org.agrona.collections.MutableInteger(); + archive.listRecordingsForUri(0, 100, CHANNEL, STREAM_ID, + (a, b, recId, c, d, e, stopPosition, f, g, h, i, j, k, l, m, n) -> { + if (stopPosition == AeronArchive.NULL_POSITION) { + active.increment(); + } + }); + return active.get(); + } + @Test public void testMissingRecordingFailsFast() { // Nothing has been recorded for this stream, so discovery must fail loudly rather than From c7aae4c8e1d49e7a78ed5b2a103d63a80f9aca58 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:28:59 -0700 Subject: [PATCH 09/10] [fix][test] Aeron archive: wait for the checkpoint instead of asserting 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. --- .../AeronArchiveSourceIntegrationTest.java | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index 4d5432a479..184ee46cf5 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -323,6 +323,21 @@ private void startSource(Map config) throws Exception { readerThread.start(); } + /** + * Waits for a checkpoint to appear. + * + *

Never assert this directly: {@link #awaitRecords} returns once records are collected, + * but the reader acknowledges after collecting and the checkpoint is written after that, on the + * poller thread. Asserting immediately races that chain and fails on slower machines — which is + * exactly how this first showed up in CI while passing locally. + */ + private void awaitCheckpoint(String because) { + Awaitility.await(because) + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .until(() -> !state.isEmpty()); + } + private void awaitRecords(int count) { Awaitility.await("source emitted " + count + " records") .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) @@ -491,7 +506,7 @@ public void testResumesFromCheckpointAfterRestart() throws Exception { readerThread = null; source.close(); source = null; - assertThat(state).as("a checkpoint should have been written").isNotEmpty(); + awaitCheckpoint("a checkpoint should have been written"); collected.clear(); // More data arrives while the source is down — exactly the window transport mode loses. @@ -516,7 +531,7 @@ public void testStartsFreshWhenNoCheckpointExists() throws Exception { awaitRecords(3); assertThat(valuesOf(collected)).containsExactly("a", "b", "c"); - assertThat(state).as("running should have produced a checkpoint").isNotEmpty(); + awaitCheckpoint("running should have produced a checkpoint"); } @Test @@ -552,7 +567,7 @@ public void testResetCheckpointReprocessesFromTheBeginning() throws Exception { readerThread = null; source.close(); source = null; - assertThat(state).as("first run should have checkpointed").isNotEmpty(); + awaitCheckpoint("first run should have checkpointed"); collected.clear(); // The escape hatch: reprocess history despite a checkpoint being present. @@ -575,7 +590,7 @@ public void testResetCheckpointDiscardsRatherThanIgnoresTheStoredPosition() thro // Deleted, then rewritten as the run progresses — so removing the flag leaves a usable // checkpoint behind rather than a stale one from before the reset. - assertThat(state).isNotEmpty(); + awaitCheckpoint("the reset run should rewrite a checkpoint"); ByteBuffer stored = state.values().iterator().next().duplicate(); stored.getLong(); assertThat(stored.getLong()).as("checkpoint rewritten by the reset run").isPositive(); From 911ecabc2462dcca240e605037d98cc33a1384f0 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:44:06 -0700 Subject: [PATCH 10/10] [fix][io] Aeron archive mode: drop the unusable concurrent-publisher workaround MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../pulsar/io/aeron/AeronSourceConfig.java | 7 ++++- .../pulsar/io/aeron/ArchivePollingRunner.java | 30 +++++++++++++------ .../AeronArchiveSourceIntegrationTest.java | 6 ++++ 3 files changed, 33 insertions(+), 10 deletions(-) diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java index 3c2c66a9e7..1afcda6865 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/AeronSourceConfig.java @@ -59,7 +59,12 @@ public class AeronSourceConfig implements Serializable { + "at-most-once: nothing can be replayed, so messages published while the " + "connector is down are lost. 'archive' replays from an Aeron Archive " + "recording and can resume after a restart. The two have materially " - + "different delivery guarantees, so the mode is explicit rather than a flag") + + "different delivery guarantees, so the mode is explicit rather than a flag. " + + "Note that 'archive' does NOT support concurrent publishers on one channel " + + "and stream: the archive records each session separately and the source " + + "replays one recording at a time, so it fails at startup if more than one " + + "recording is active. Give each publisher a distinct channel or streamId, " + + "or use 'transport', which does receive from every publication") private String mode = MODE_TRANSPORT; @FieldDoc( diff --git a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java index 95c6fa836f..caf0154dfd 100644 --- a/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java +++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java @@ -102,11 +102,16 @@ * *

One active recording at a time

* - *

Transport mode receives from every publication on a channel and stream. Archive mode cannot: - * the archive records each session separately, so concurrent publishers produce concurrent - * recordings and this replays one at a time. Rather than silently dropping the others, it fails at - * {@code open()} when more than one is active. Give each publisher a distinct channel or stream, - * or pin {@code recordingId} and run a source per recording. + *

Transport mode receives from every publication on a channel and stream. Archive mode does + * not support concurrent publishers: the archive records each session separately, so they + * produce concurrent recordings and this replays one at a time. Rather than silently dropping the + * others, it fails at {@code open()} when more than one is active. + * + *

The remedy is a distinct channel or stream id per publisher, or {@code mode: transport} where + * multi-publisher delivery works. Pinning {@code recordingId} and running a source per recording is + * not a workaround: the guard runs before the start position is resolved so it rejects + * pinned configurations too, and pinned runners auto-advance on rotation and would walk into each + * other's recordings. * *

No record sequence is set. An earlier revision exposed the archive position as * {@link Record#getRecordSequence()} so broker deduplication could give effectively-once, but @@ -229,13 +234,20 @@ private void rejectConcurrentRecordings() { } while (matched == LIST_PAGE_SIZE); if (active.get() > 1) { + // Deliberately does NOT suggest pinning 'recordingId' and running a source per + // recording. This 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. Suggesting a workaround that + // cannot work is worse than stating the limitation. throw new IllegalStateException( "Archive mode found " + active.get() + " concurrently active recordings for " + "channel '" + config.getChannel() + "' streamId " - + config.getStreamId() + " (recordingIds: " + ids + "). It replays one " - + "recording at a time, so the others would never be replayed. Use a " - + "distinct channel or streamId per publisher, or set 'recordingId' " - + "explicitly and run one source per recording."); + + config.getStreamId() + " (recordingIds: " + ids + "). Concurrent " + + "publishers on one channel and stream are not supported in archive " + + "mode: the archive records each session separately and this replays " + + "one recording at a time, so the others would never be replayed. " + + "Give each publisher a distinct channel or streamId, or use " + + "'mode: transport', which does receive from every publication."); } } diff --git a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java index 184ee46cf5..db2bda94a7 100644 --- a/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java +++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java @@ -858,6 +858,12 @@ public void testConcurrentActiveRecordingsAreRejected() throws Exception { } catch (Exception e) { assertThat(e).isInstanceOf(IllegalStateException.class); assertThat(e).hasMessageContaining("concurrently active recordings"); + assertThat(e).hasMessageContaining("not supported in archive mode"); + // The message must not offer a workaround that cannot work: this guard runs + // before the start position is resolved, so a pinned recordingId is rejected too. + assertThat(e.getMessage()) + .as("must not suggest pinning recordingId, which this guard rejects anyway") + .doesNotContain("set 'recordingId'"); } finally { closeQuietly(bad); }