Manual Offset Commit

[kriti-sc]

The sink runtime currently configures consumers with AutoCommit::When(AutoCommitWhen::PollingMessages) (runtime/src/sink.rs:421), meaning offsets are committed when messages are polled — before they are processed. A failed batch is silently lost.

Sink must have an option to decide exactly when it is safe to advance the offset, as only the sink knows when a write is truly durable from the target's perspective.

Proposed Solution

Offset information is already available to the sink. MessagesMetadata (passed to Sink::consume) contains partition_id and current_offset. Each ConsumedMessage carries its own offset. What is missing is a mechanism to actually call commit.

The runtime passes a CommitCallback function pointer into the sink at open time, analogous to how LogCallback is already passed. The sink stores it and calls commit_callback(partition_id, offset) when its write to the target is durable. The runtime switches to AutoCommit::Disabled and trusts the sink to drive commits.

Note: The above mentioned CommitCallback mechanism is a breaking change but a non-breaking alternative is proposed below.

Implementation

Three pieces of work are required.

1. Configurable Commit Mode (OffsetCommitMode)

Add a new field to SinkConfig (runtime/src/configs/connectors.rs:106):

#[serde(default)]
pub offset_commit: OffsetCommitMode,

With the following variants:

• Auto — default; current behaviour, AutoCommit::When(AutoCommitWhen::PollingMessages); existing sinks work unchanged
• SinkControlled — AutoCommit::Disabled; runtime passes a CommitCallback to the sink at open time, sink drives all offset commits

setup_sink_consumers (runtime/src/sink.rs:379) reads this field and sets the consumer's AutoCommit accordingly:

let auto_commit = match config.offset_commit {
    OffsetCommitMode::Auto           => AutoCommit::When(AutoCommitWhen::PollingMessages),
    OffsetCommitMode::SinkControlled => AutoCommit::Disabled,
};

2. CommitCallback in the ABI

Add to the SDK:

pub type CommitCallback = extern "C" fn(plugin_id: u32, stream_id: u32, topic_id: u32, partition_id: u32, offset: u64) -> i32;

The SDK's SinkContainer stores the callback and exposes it to the sink instance via a CommitHandle wrapper. Two options for how the callback is delivered:

Option A: Extend iggy_sink_open (breaking)

iggy_sink_open gains a commit_callback: CommitCallback parameter. All existing plugins break at the ABI level and must be recompiled. The runtime always passes either the real callback (when SinkControlled) or a no-op stub (when Auto).

Option B: Separate iggy_sink_set_commit_callback symbol (non-breaking)

A new optional symbol is added:

extern "C" fn iggy_sink_set_commit_callback(plugin_id: u32, callback: CommitCallback) -> i32;

The runtime calls it after iggy_sink_open, only when SinkControlled. Existing plugins that do not export this symbol continue to load and run unchanged — the runtime detects the symbol's absence via dlopen2 and treats it as Auto mode. Only plugins that opt into SinkControlled in config need to export the new symbol.

Option B is preferred as it preserves backwards compatibility with all existing plugins.

3. Global Offset Sender Registry

The CommitCallback is extern "C" — a bare function pointer that cannot capture state. It needs to route a commit call from the plugin back to the correct IggyConsumer instance in the runtime. The solution follows the same pattern as SOURCE_SENDERS (runtime/src/source.rs:52), which solves the identical problem for sources:

pub static SINK_COMMIT_SENDERS: Lazy<DashMap<(u32, u32, u32, u32), flume::Sender<u64>>> =
    Lazy::new(DashMap::new);

The registry is keyed by (plugin_id, stream_id, topic_id, partition_id). Each part is necessary: a plugin can have multiple consumers across streams and topics, and partition IDs are only unique within a topic. The sink has all four values at hand — stream_id and topic_id from TopicMetadata, partition_id from MessagesMetadata, and plugin_id baked into the CommitHandle.

The runtime, for each consumer created under SinkControlled:

1. Creates the consumer
2. Inserts its store_offset_sender into SINK_COMMIT_SENDERS keyed by (plugin_id, stream_id, topic_id, partition_id)
3. Passes the static commit_callback into iggy_sink_open

Considerations:

• Cleanup — entries must be removed on sink shutdown or restart, same as SOURCE_SENDERS::cleanup_sender; stale entries do not cause correctness issues (the channel will be closed and sends will fail) but should be cleaned up to avoid unbounded growth
• Channel closed — if the consumer shuts down before the sink calls commit, the send will fail; the callback returns -1 so the sink can detect and handle the failure

[spetz]

I like the idea, and for starters, 2 modes could be enough, e.g., AfterPolling (current one) and AfterConsuming - not sure about these names, but the more explicit these are, I think the better for the users. However, I'm not sure if we need an additional callback mechanism.

Actually, why not just make use of the existing mut consumer: IggyConsumer within consume_messages() and, depending on the mode, either do nothing or invoke store_offset() after process_messages has completed successfully?