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..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 {
@@ -44,10 +49,26 @@ 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";
+
+ /**
+ * 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 Outcome outcome;
/**
* @param value the reassembled payload; must already be a copy owned by this record
@@ -55,9 +76,18 @@ 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 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
+ */
+ AeronRecord(byte[] value, String key, Map properties, Outcome outcome) {
this.value = value;
this.key = key;
this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
+ this.outcome = outcome;
}
@Override
@@ -77,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/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..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
@@ -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,30 @@ 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. "
+ + "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(
required = true,
defaultValue = "",
@@ -96,6 +124,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 +243,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..08aa9a999e
--- /dev/null
+++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchiveCheckpoint.java
@@ -0,0 +1,169 @@
+/*
+ * 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);
+ }
+ }
+
+ /**
+ * 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 {
+ buffer = sourceContext.getState(key);
+ } catch (Exception 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()));
+ }
+
+ /**
+ * 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() {
+ Exception deleteFailure = null;
+ try {
+ sourceContext.deleteState(key);
+ } catch (Exception 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;
+ }
+
+ // 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 "
+ + "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);
+ }
+ }
+
+ 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..caf0154dfd
--- /dev/null
+++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/ArchivePollingRunner.java
@@ -0,0 +1,643 @@
+/*
+ * 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.Image;
+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:
+ *
+ *
+ * - {@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.
+ *
- A stored checkpoint, if one exists.
+ *
- {@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
+ *
+ * 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 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.
+ *
+ *
One active recording at a time
+ *
+ * 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
+ * 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 {
+
+ 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;
+
+ /** 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;
+ private final SourceContext sourceContext;
+ 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;
+ private long currentPosition;
+ private long recordsSinceCheckpoint;
+ private long lastCheckpointedPosition = Long.MIN_VALUE;
+ private long lastCheckpointedRecording = Aeron.NULL_VALUE;
+ private long lastRotationStallLogNanos;
+
+ 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();
+ 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) {
+ // 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 + "). 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.");
+ }
+ }
+
+ /**
+ * 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);
+ // 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);
+ }
+
+ /**
+ * 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);
+ 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);
+ }
+ } else if (recordingId > after
+ && (best.get() == Aeron.NULL_VALUE || recordingId < best.get())) {
+ best.set(recordingId);
+ }
+ });
+ // 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) {
+ 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();
+ }
+
+ /**
+ * 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);
+ 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();
+ }
+
+ /**
+ * 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);
+ } 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) {
+ 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.
+ try {
+ writeCheckpointIfAdvanced();
+ } 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);
+ }
+ }
+
+ /**
+ * 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;
+ try (Subscription replay = archive.replay(
+ currentRecordingId, currentPosition, length,
+ config.getReplayChannel(), config.getReplayStreamId())) {
+
+ // "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;
+ // 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;
+ final Image image = replay.imageAtIndex(0);
+ imagePosition = Math.max(imagePosition, image.position());
+ if (image.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);
+ }
+
+ // 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;
+ }
+ }
+ }
+
+ /**
+ * 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
+ }
+
+ // 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
+ }
+
+ LOG.info("Recording {} complete at {}; advancing to recording {}",
+ currentRecordingId, stop, next);
+ currentRecordingId = next;
+ // 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);
+ // A partially assembled message cannot continue into a different recording.
+ assembler = newAssembler();
+ recordsSinceCheckpoint = 0;
+ writeCheckpointIfAdvanced();
+ 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;
+ 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);
+ }
+
+ @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);
+
+ // 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()) {
+ writeCheckpointIfAdvanced();
+ }
+ }
+
+ /** 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.
+ if (!commitTracker.hasCommitted()) {
+ 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) {
+ // 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, committed, 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/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..136f6a73c2
--- /dev/null
+++ b/aeron/src/main/java/org/apache/pulsar/io/aeron/CommitTracker.java
@@ -0,0 +1,119 @@
+/*
+ * 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.
+ *
+ *
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
+ * 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
new file mode 100644
index 0000000000..db2bda94a7
--- /dev/null
+++ b/aeron/src/test/java/org/apache/pulsar/io/aeron/AeronArchiveSourceIntegrationTest.java
@@ -0,0 +1,903 @@
+/*
+ * 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.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;
+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;
+ /** 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;
+
+ @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<>();
+ ackRecords = true;
+ failRecords = false;
+ 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();
+ });
+ // 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.
+ 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) {
+ 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(publicationChannel, 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()) {
+ 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 (failRecords) {
+ record.fail();
+ } else if (ackRecords) {
+ record.ack();
+ }
+ }
+ } catch (Exception e) {
+ // Interrupted at teardown.
+ }
+ }, "aeron-archive-it-reader");
+ readerThread.setDaemon(true);
+ 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)
+ .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 testNoRecordSequenceIsExposed() throws Exception {
+ recordMessages(List.of("alpha", "beta", "gamma"));
+
+ startSource(archiveConfig());
+ awaitRecords(3);
+
+ // 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(positions).isSorted();
+ assertThat(positions.get(positions.size() - 1)).isGreaterThan(positions.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 =
+ Long.parseLong(collected.get(0).getProperties().get(AeronRecord.PROP_POSITION));
+
+ 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;
+ awaitCheckpoint("a checkpoint should have been written");
+ 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");
+ awaitCheckpoint("running should have produced a checkpoint");
+ }
+
+ @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;
+ awaitCheckpoint("first run should have checkpointed");
+ 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.
+ 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();
+ }
+
+ @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 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 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 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 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");
+ 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);
+ }
+ } 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
+ // 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..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
@@ -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();
@@ -105,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/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..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
@@ -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();
@@ -104,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();
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" }