feat(connectors): let sinks control when offsets are committed

Cargo.toml

@@ -56,6 +56,7 @@ members = [
     "core/cpu_allocation",
     "core/harness_derive",
     "core/integration",
+    "core/integration/fixtures/test_sink",
     "core/journal",
     "core/message_bus",
     "core/metadata",

core/connectors/runtime/src/configs/connectors.rs

@@ -71,6 +71,14 @@ impl ConnectorConfig {
     }
 }
 
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OffsetCommitMode {
+    #[default]
+    AfterPolling,
+    AfterConsuming,
+}
+
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct CreateSinkConfig {
     pub enabled: bool,
@@ -84,6 +92,8 @@ pub struct CreateSinkConfig {
     pub verbose: bool,
     #[serde(default)]
     pub benchmark: bool,
+    #[serde(default)]
+    pub offset_commit: OffsetCommitMode,
 }
 
 impl CreateSinkConfig {
@@ -100,6 +110,7 @@ impl CreateSinkConfig {
             plugin_config: self.plugin_config.clone(),
             verbose: self.verbose,
             benchmark: self.benchmark,
+            offset_commit: self.offset_commit,
         }
     }
 }
@@ -122,6 +133,9 @@ pub struct SinkConfig {
     pub verbose: bool,
     #[serde(default)]
     pub benchmark: bool,
+    #[serde(default)]
+    #[config_env(leaf)]
+    pub offset_commit: OffsetCommitMode,
 }
 
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
@@ -411,3 +425,55 @@ impl ConnectorsConfig {
         &self.sources
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use configs::ConfigEnvMappings;
+
+    const MINIMAL_SINK_TOML: &str = r#"
+        key = "test"
+        enabled = true
+        version = 1
+        name = "test sink"
+        path = "libtest_sink"
+        streams = []
+    "#;
+
+    #[test]
+    fn given_sink_config_without_offset_commit_when_deserialized_should_default_to_after_polling() {
+        let config: SinkConfig = toml::from_str(MINIMAL_SINK_TOML).expect("failed to parse config");
+        assert_eq!(config.offset_commit, OffsetCommitMode::AfterPolling);
+    }
+
+    #[test]
+    fn given_sink_config_with_after_consuming_when_deserialized_should_parse_mode() {
+        let toml = format!("{MINIMAL_SINK_TOML}\noffset_commit = \"after_consuming\"\n");
+        let config: SinkConfig = toml::from_str(&toml).expect("failed to parse config");
+        assert_eq!(config.offset_commit, OffsetCommitMode::AfterConsuming);
+    }
+
+    #[test]
+    fn given_unknown_offset_commit_mode_when_deserialized_should_fail() {
+        let toml = format!("{MINIMAL_SINK_TOML}\noffset_commit = \"after_flushing\"\n");
+        let result: Result<SinkConfig, _> = toml::from_str(&toml);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn given_create_sink_config_when_converted_should_carry_offset_commit() {
+        let create = CreateSinkConfig {
+            offset_commit: OffsetCommitMode::AfterConsuming,
+            ..CreateSinkConfig::default()
+        };
+        let config = create.to_sink_config("test", 1);
+        assert_eq!(config.offset_commit, OffsetCommitMode::AfterConsuming);
+    }
+
+    #[test]
+    fn given_sink_config_env_mappings_should_expose_offset_commit_as_leaf() {
+        let mapping = <SinkConfig as ConfigEnvMappings>::find_by_config_path("offset_commit")
+            .expect("offset_commit is not exposed as an env var mapping");
+        assert!(mapping.env_name.ends_with("OFFSET_COMMIT"));
+    }
+}

core/connectors/runtime/src/error.rs

@@ -27,6 +27,8 @@ pub enum RuntimeError {
     FailedToSerializeMessagesMetadata,
     #[error("Failed to serialize raw messages")]
     FailedToSerializeRawMessages,
+    #[error("Sink connector with ID: {0} rejected the batch with code: {1}")]
+    SinkRejectedBatch(u32, i32),
     #[error("Connector SDK error")]
     ConnectorSdkError(#[from] iggy_connector_sdk::Error),
     /// A classified state-store failure while loading an enabled source's

core/connectors/runtime/src/main.rs

@@ -18,7 +18,7 @@
 use crate::configs::connectors::{ConnectorsConfigProvider, create_connectors_config_provider};
 use ::configs::ConfigProvider;
 use clap::Parser;
-use configs::connectors::ConfigFormat;
+use configs::connectors::{ConfigFormat, OffsetCommitMode};
 use configs::runtime::ConnectorsRuntimeConfig;
 use dlopen2::wrapper::{Container, WrapperApi};
 use dotenvy::dotenv;
@@ -421,6 +421,7 @@ struct SinkConnectorPlugin {
     error: Option<String>,
     verbose: bool,
     benchmark: bool,
+    offset_commit: OffsetCommitMode,
 }
 
 struct SinkConnectorConsumer {

core/connectors/runtime/src/manager/sink.rs

@@ -214,6 +214,7 @@ impl SinkManager {
             callback,
             config.verbose,
             config.benchmark,
+            config.offset_commit,
             metrics,
             context.clone(),
         );

core/connectors/runtime/src/sink.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use crate::benchmark;
-use crate::configs::connectors::SinkConfig;
+use crate::configs::connectors::{OffsetCommitMode, SinkConfig};
 use crate::context::RuntimeContext;
 use crate::log::LOG_CALLBACK;
 use crate::metrics::{Metrics, SinkLabels};
@@ -147,6 +147,7 @@ pub async fn init(
             error: init_error.clone(),
             verbose: config.verbose,
             benchmark: config.benchmark,
+            offset_commit: config.offset_commit,
         });
 
         if let Some(error) = init_error {
@@ -229,6 +230,7 @@ pub fn consume(
                 sink.callback,
                 plugin.verbose,
                 plugin.benchmark,
+                plugin.offset_commit,
                 &context.metrics,
                 context.clone(),
             );
@@ -251,6 +253,7 @@ pub(crate) fn spawn_consume_tasks(
     callback: ConsumeCallback,
     verbose: bool,
     benchmark: bool,
+    offset_commit: OffsetCommitMode,
     metrics: &Arc<Metrics>,
     context: Arc<RuntimeContext>,
 ) -> (watch::Sender<()>, Vec<JoinHandle<()>>) {
@@ -267,6 +270,7 @@ pub(crate) fn spawn_consume_tasks(
         let plugin_key = plugin_key.to_string();
         let metrics = metrics.clone();
         let shutdown_rx = shutdown_rx.clone();
+        let shutdown_tx = shutdown_tx.clone();
         let context = context.clone();
         let labels = labels.clone();
         let handle = tokio::spawn(async move {
@@ -279,6 +283,7 @@ pub(crate) fn spawn_consume_tasks(
                 consumer,
                 verbose,
                 benchmark,
+                offset_commit,
                 &plugin_key,
                 &metrics,
                 &labels,
@@ -294,6 +299,11 @@ pub(crate) fn spawn_consume_tasks(
                     .sinks
                     .set_error(&plugin_key, &error.to_string())
                     .await;
+                // The instance owns the target connection, so one topic's
+                // failure condemns the rest. Stopping them here keeps the
+                // failure domain the same as the recovery domain: the whole
+                // connector goes down, and `restart_connector` brings it back.
+                let _ = shutdown_tx.send(());
             }
         });
         task_handles.push(handle);
@@ -311,6 +321,7 @@ pub(crate) async fn consume_messages(
     mut consumer: IggyConsumer,
     verbose: bool,
     benchmark: bool,
+    offset_commit: OffsetCommitMode,
     plugin_key: &str,
     metrics: &Arc<Metrics>,
     labels: &SinkLabels,
@@ -389,6 +400,11 @@ pub(crate) async fn consume_messages(
         // Total always records; sub-stages only on success (no 0-sample skew).
         metrics.observe_stage_with_labels(&labels.stage_total, elapsed);
 
+        let consume_result = match &result {
+            Ok(timing) => timing.consume_result,
+            Err(_) => 0,
+        };
+
         let (processed_count, decode_us, prepare_us, ffi_us) = match &result {
             Ok(timing) => {
                 let prepare_elapsed = elapsed
@@ -430,6 +446,33 @@ pub(crate) async fn consume_messages(
             return Err(error);
         }
 
+        if consume_result != 0 {
+            error!(
+                "Sink connector with ID: {plugin_id} rejected {messages_count} messages from \
+                 stream: {}, topic: {}, partition ID: {partition_id} with code: {consume_result}",
+                topic_metadata.stream, topic_metadata.topic,
+            );
+            metrics.inc_errors_with_labels(&labels.counter);
+            // A rejection means the target is unusable, not that this batch is
+            // bad - a sink drops bad records itself and returns success. Both
+            // modes stop: continuing would hand every later batch to the same
+            // failing target, and under `AfterPolling` each of those is already
+            // committed at poll time, so the topic would drain into nothing.
+            return Err(RuntimeError::SinkRejectedBatch(plugin_id, consume_result));
+        }
+
+        if offset_commit == OffsetCommitMode::AfterConsuming
+            && let Err(error) = consumer
+                .store_offset(message_offset, Some(partition_id))
+                .await
+        {
+            error!(
+                "Failed to store offset: {message_offset} for partition ID: {partition_id}, \
+                 sink connector with ID: {plugin_id}. {error}",
+            );
+            return Err(error.into());
+        }
+
         metrics.inc_messages_processed_with_labels(&labels.counter, processed_count as u64);
         if verbose {
             info!(
@@ -502,6 +545,11 @@ pub(crate) async fn setup_sink_consumers(
         vec![]
     };
 
+    let auto_commit = match config.offset_commit {
+        OffsetCommitMode::AfterPolling => AutoCommit::When(AutoCommitWhen::PollingMessages),
+        OffsetCommitMode::AfterConsuming => AutoCommit::Disabled,
+    };
+
     let mut consumers = Vec::new();
     for stream in config.streams.iter() {
         let poll_interval = IggyDuration::from_str(
@@ -519,7 +567,7 @@ pub(crate) async fn setup_sink_consumers(
         for topic in stream.topics.iter() {
             let mut consumer = iggy_client
                 .consumer_group(consumer_group, &stream.stream, topic)?
-                .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
+                .auto_commit(auto_commit)
                 .create_consumer_group_if_not_exists()
                 .auto_join_consumer_group()
                 .polling_strategy(PollingStrategy::next())
@@ -737,7 +785,7 @@ async fn process_messages(
     })?;
 
     let ffi_start = Instant::now();
-    (consume)(
+    let consume_result = (consume)(
         plugin_id,
         topic_meta.as_ptr(),
         topic_meta.len(),
@@ -752,11 +800,15 @@ async fn process_messages(
         processed_count,
         decode_elapsed,
         ffi_elapsed,
+        consume_result,
     })
 }
 
 struct SinkBatchTiming {
     processed_count: usize,
     decode_elapsed: Duration,
     ffi_elapsed: Duration,
+    /// Plugin's `iggy_sink_consume` return code: 0 on success, non-zero when
+    /// the sink rejected the batch.
+    consume_result: i32,
 }

core/integration/fixtures/test_sink/Cargo.toml

@@ -0,0 +1,41 @@
+# 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]
+name = "iggy_connector_test_sink"
+version = "0.5.0-edge.4"
+description = "Sink connector plugin used only by the Iggy integration test suite to drive controllable success and failure behaviour."
+edition = "2024"
+license = "Apache-2.0"
+publish = false
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[lints]
+workspace = true