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

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)

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}.
  *
- * <p>{@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.
+ * <p>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.
+ *
+ * <p>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<byte[]> {
 
@@ -44,20 +49,45 @@ public class AeronRecord implements Record<byte[]> {
     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.
+     *
+     * <p>Archive mode needs this: the framework publishes and acknowledges <em>after</em>
+     * {@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<String, String> properties;
+    private final Outcome outcome;
 
     /**
      * @param value      the reassembled payload; must already be a copy owned by this record
      * @param key        the record key, or null for none
      * @param properties Aeron metadata; copied defensively
      */
     public AeronRecord(byte[] value, String key, Map<String, String> 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<String, String> 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<String, String> 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();
+        }
     }
 }

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()}.
  *
- * <p><b>Delivery semantics: at-most-once.</b> 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.
+ * <p><b>Delivery semantics depend on the configured mode, and they differ materially.</b>
+ *
+ * <ul>
+ *   <li><b>{@code transport}</b> (default) reads the live Aeron stream and is
+ *       <b>at-most-once</b>. 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.
+ *   <li><b>{@code archive}</b> replays from an Aeron Archive recording, so a position in durable
+ *       storage exists and data missed while the connector was down can be re-read. Records carry
+ *       the archive position as their record sequence, which broker deduplication can use.
+ * </ul>
+ *
+ * <p>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.
  *
  * <p>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<byte[]> {
     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<String, Object> 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();

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<String> 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<String, Object> 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.
+     *
+     * <p>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<String> 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 + "'");
+        }
     }
 }