From 3b2a1448ed74f2af4d514d4f740c3a08d4efd30b Mon Sep 17 00:00:00 2001 From: kriti-sc Date: Sat, 22 Aug 2026 18:20:56 +0530 Subject: [PATCH] add support for sinks to drive offset commits --- Cargo.lock | 12 + Cargo.toml | 1 + .../runtime/src/configs/connectors.rs | 66 +++++ core/connectors/runtime/src/error.rs | 2 + core/connectors/runtime/src/main.rs | 3 +- core/connectors/runtime/src/manager/sink.rs | 1 + core/connectors/runtime/src/sink.rs | 58 +++- .../integration/fixtures/test_sink/Cargo.toml | 41 +++ .../integration/fixtures/test_sink/src/lib.rs | 130 +++++++++ .../tests/connectors/runtime/mod.rs | 1 + .../tests/connectors/runtime/offset_commit.rs | 268 ++++++++++++++++++ .../offset_commit_after_consuming.toml | 20 ++ .../test_sink.toml | 34 +++ ...offset_commit_after_consuming_failing.toml | 20 ++ .../test_sink.toml | 35 +++ .../offset_commit_after_polling_failing.toml | 20 ++ .../test_sink.toml | 34 +++ .../runtime/offset_commit_multi_topic.toml | 21 ++ .../test_sink.toml | 36 +++ 19 files changed, 799 insertions(+), 4 deletions(-) create mode 100644 core/integration/fixtures/test_sink/Cargo.toml create mode 100644 core/integration/fixtures/test_sink/src/lib.rs create mode 100644 core/integration/tests/connectors/runtime/offset_commit.rs create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_consuming.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_consuming_config/test_sink.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing_config/test_sink.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_polling_failing.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_after_polling_failing_config/test_sink.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_multi_topic.toml create mode 100644 core/integration/tests/connectors/runtime/offset_commit_multi_topic_config/test_sink.toml diff --git a/Cargo.lock b/Cargo.lock index a08a49adb9..9baf0ecd30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7222,6 +7222,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_test_sink" +version = "0.5.0-edge.4" +dependencies = [ + "async-trait", + "dashmap", + "iggy_connector_sdk", + "serde", + "tokio", + "tracing", +] + [[package]] name = "iggy_examples" version = "0.0.6" diff --git a/Cargo.toml b/Cargo.toml index 7cefb5fdf7..3cb66f4ea0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,6 +55,7 @@ members = [ "core/cpu_allocation", "core/harness_derive", "core/integration", + "core/integration/fixtures/test_sink", "core/journal", "core/message_bus", "core/metadata", diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs index 24647e8703..ce7ad0cc0f 100644 --- a/core/connectors/runtime/src/configs/connectors.rs +++ b/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 = 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 = ::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")); + } +} diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs index a8d0ba7647..afc14c5e64 100644 --- a/core/connectors/runtime/src/error.rs +++ b/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), #[error("Iggy client error")] diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 25677d3bd2..06937af385 100644 --- a/core/connectors/runtime/src/main.rs +++ b/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; @@ -423,6 +423,7 @@ struct SinkConnectorPlugin { error: Option, verbose: bool, benchmark: bool, + offset_commit: OffsetCommitMode, } struct SinkConnectorConsumer { diff --git a/core/connectors/runtime/src/manager/sink.rs b/core/connectors/runtime/src/manager/sink.rs index 270d442d9b..5b2cb37bc5 100644 --- a/core/connectors/runtime/src/manager/sink.rs +++ b/core/connectors/runtime/src/manager/sink.rs @@ -214,6 +214,7 @@ impl SinkManager { callback, config.verbose, config.benchmark, + config.offset_commit, metrics, context.clone(), ); diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs index 7a17724510..03af34b951 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/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, context: Arc, ) -> (watch::Sender<()>, Vec>) { @@ -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, 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,6 +800,7 @@ async fn process_messages( processed_count, decode_elapsed, ffi_elapsed, + consume_result, }) } @@ -759,4 +808,7 @@ 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, } diff --git a/core/integration/fixtures/test_sink/Cargo.toml b/core/integration/fixtures/test_sink/Cargo.toml new file mode 100644 index 0000000000..efa1aab191 --- /dev/null +++ b/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 diff --git a/core/integration/fixtures/test_sink/src/lib.rs b/core/integration/fixtures/test_sink/src/lib.rs new file mode 100644 index 0000000000..59a9799662 --- /dev/null +++ b/core/integration/fixtures/test_sink/src/lib.rs @@ -0,0 +1,130 @@ +// 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. + +//! Sink plugin with configurable failure behaviour, for integration tests that +//! need to observe what the runtime does when a sink rejects a batch. +//! +//! Not shipped: this crate exists only to back tests under +//! `core/integration/tests/connectors/`. + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use serde::{Deserialize, Serialize}; +use tokio::sync::Mutex; +use tracing::{error, info}; + +sink_connector!(TestSink); + +#[derive(Debug)] +struct State { + batches_consumed: usize, +} + +#[derive(Debug)] +pub struct TestSink { + id: u32, + fail_after_batches: Option, + reject_topics: Vec, + state: Mutex, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct TestSinkConfig { + /// Accept this many batches, then reject every batch after. `None` accepts + /// everything; `Some(0)` rejects the first batch. + fail_after_batches: Option, + /// Reject only batches from these topics. Lets a multi-topic sink fail on + /// one topic while still accepting the others, so a test can tell a stopped + /// task apart from one that ran and was rejected. + reject_topics: Option>, +} + +impl TestSink { + pub fn new(id: u32, config: TestSinkConfig) -> Self { + TestSink { + id, + fail_after_batches: config.fail_after_batches, + reject_topics: config.reject_topics.unwrap_or_default(), + state: Mutex::new(State { + batches_consumed: 0, + }), + } + } +} + +#[async_trait] +impl Sink for TestSink { + async fn open(&mut self) -> Result<(), Error> { + info!( + "Opened test sink connector with ID: {}, fail after batches: {:?}, reject topics: {:?}", + self.id, self.fail_after_batches, self.reject_topics + ); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec, + ) -> Result<(), Error> { + let mut state = self.state.lock().await; + let batch_index = state.batches_consumed; + let topic_selected = + self.reject_topics.is_empty() || self.reject_topics.contains(&topic_metadata.topic); + let should_fail = topic_selected + && self + .fail_after_batches + .is_some_and(|threshold| batch_index >= threshold); + if !should_fail { + state.batches_consumed += 1; + } + drop(state); + + if should_fail { + error!( + "Test sink with ID: {} rejecting batch: {batch_index} of {} messages, stream: {}, topic: {}, partition: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + ); + return Err(Error::CannotStoreData(format!( + "test sink configured to reject batches from index {batch_index}" + ))); + } + + info!( + "Test sink with ID: {} accepted batch: {batch_index} of {} messages, stream: {}, topic: {}, partition: {}, last offset: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages.last().map(|message| message.offset).unwrap_or(0), + ); + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + info!("Test sink connector with ID: {} is closed.", self.id); + Ok(()) + } +} diff --git a/core/integration/tests/connectors/runtime/mod.rs b/core/integration/tests/connectors/runtime/mod.rs index 86294e8daa..4f01d01104 100644 --- a/core/integration/tests/connectors/runtime/mod.rs +++ b/core/integration/tests/connectors/runtime/mod.rs @@ -17,3 +17,4 @@ mod benchmark; mod error_isolation; +mod offset_commit; diff --git a/core/integration/tests/connectors/runtime/offset_commit.rs b/core/integration/tests/connectors/runtime/offset_commit.rs new file mode 100644 index 0000000000..404f65b0b4 --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit.rs @@ -0,0 +1,268 @@ +// 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. + +//! Offset commit mode tests for sink connectors. +//! +//! The runtime commits consumer offsets in one of two places, selected by the +//! `offset_commit` key on a sink config: +//! * `after_polling` (default) - the SDK auto-commits when a message is +//! polled, before the sink has seen it. A sink that rejects the batch still +//! leaves the offset advanced, so those messages are never redelivered. +//! * `after_consuming` - auto-commit is disabled and the runtime stores the +//! offset only once the sink accepts the batch. +//! +//! Each test drives `test_sink`, a fixture plugin whose `fail_after_batches` +//! config decides whether it accepts or rejects batches, then reads the stored +//! consumer group offset back from the server. + +use iggy::prelude::{ + Consumer, ConsumerOffsetClient, Identifier, IggyMessage, MessageClient, Partitioning, +}; +use iggy_connector_sdk::api::{ConnectorStatus, SinkInfoResponse}; +use integration::harness::seeds; +use integration::harness::{TestHarness, seeds::names}; +use integration::iggy_harness; +use reqwest::Client; +use std::time::Duration; +use tokio::time::sleep; + +const MESSAGE_COUNT: usize = 10; +const SINK_KEY: &str = "offset_commit_sink"; +const OFFSET_POLL_ATTEMPTS: u32 = 50; +const OFFSET_POLL_INTERVAL: Duration = Duration::from_millis(200); +/// How long to wait before asserting an offset stayed absent. Long enough that +/// a commit the runtime was going to make would already have landed. +const NO_COMMIT_OBSERVATION_WINDOW: Duration = Duration::from_secs(3); + +async fn send_test_messages(harness: &TestHarness) { + send_test_messages_to(harness, names::TOPIC).await; +} + +async fn send_test_messages_to(harness: &TestHarness, topic: &str) { + let client = harness.root_client().await.expect("failed to build client"); + let stream_id: Identifier = names::STREAM.try_into().unwrap(); + let topic_id: Identifier = topic.try_into().unwrap(); + + let mut messages: Vec = (0..MESSAGE_COUNT) + .map(|index| { + IggyMessage::builder() + .id((index + 1) as u128) + .payload(format!(r#"{{"index":{index}}}"#).into()) + .build() + .expect("failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("failed to send messages"); +} + +async fn stored_offset(harness: &TestHarness, consumer_group: &str) -> Option { + stored_offset_for(harness, consumer_group, names::TOPIC).await +} + +async fn stored_offset_for( + harness: &TestHarness, + consumer_group: &str, + topic: &str, +) -> Option { + let client = harness.root_client().await.expect("failed to build client"); + let stream_id: Identifier = names::STREAM.try_into().unwrap(); + let topic_id: Identifier = topic.try_into().unwrap(); + let group_id: Identifier = consumer_group.try_into().unwrap(); + + client + .get_consumer_offset(&Consumer::group(group_id), &stream_id, &topic_id, None) + .await + .expect("failed to query consumer offset") + .map(|info| info.stored_offset) +} + +/// Waits for the sink to report `status`, returning the last status seen so a +/// failure reports what the connector actually settled on. +async fn wait_for_sink_status(harness: &TestHarness, status: ConnectorStatus) -> ConnectorStatus { + let api_address = harness + .connectors_runtime() + .expect("connector runtime should be available") + .http_url(); + let http_client = Client::new(); + let mut last = ConnectorStatus::Running; + + for _ in 0..OFFSET_POLL_ATTEMPTS { + let sinks: Vec = http_client + .get(format!("{api_address}/sinks")) + .send() + .await + .expect("failed to query /sinks") + .json() + .await + .expect("failed to parse sinks"); + last = sinks + .iter() + .find(|sink| sink.key == SINK_KEY) + .expect("sink should be reported") + .status; + if last == status { + return last; + } + sleep(OFFSET_POLL_INTERVAL).await; + } + last +} + +/// Waits for the consumer group offset to reach `expected`, returning the last +/// value seen so a failure reports what the offset actually was. +async fn wait_for_stored_offset( + harness: &TestHarness, + consumer_group: &str, + expected: u64, +) -> Option { + let mut last = None; + for _ in 0..OFFSET_POLL_ATTEMPTS { + last = stored_offset(harness, consumer_group).await; + if last == Some(expected) { + return last; + } + sleep(OFFSET_POLL_INTERVAL).await; + } + last +} + +#[iggy_harness( + server(connectors_runtime( + config_path = "tests/connectors/runtime/offset_commit_after_consuming.toml" + )), + seed = seeds::connector_stream +)] +async fn given_after_consuming_when_sink_accepts_batch_should_advance_offset( + harness: &TestHarness, +) { + send_test_messages(harness).await; + + let last_offset = (MESSAGE_COUNT - 1) as u64; + let offset = + wait_for_stored_offset(harness, "offset_commit_after_consuming", last_offset).await; + + assert_eq!( + offset, + Some(last_offset), + "with offset_commit = after_consuming the runtime should store the last consumed offset \ + once the sink accepts the batch" + ); +} + +#[iggy_harness( + server(connectors_runtime( + config_path = "tests/connectors/runtime/offset_commit_after_consuming_failing.toml" + )), + seed = seeds::connector_stream +)] +async fn given_after_consuming_when_sink_rejects_batch_should_not_advance_offset( + harness: &TestHarness, +) { + send_test_messages(harness).await; + + let status = wait_for_sink_status(harness, ConnectorStatus::Error).await; + assert_eq!( + status, + ConnectorStatus::Error, + "a rejected batch must stop the sink rather than hand the next batch to the same \ + failing target" + ); + + sleep(NO_COMMIT_OBSERVATION_WINDOW).await; + let offset = stored_offset(harness, "offset_commit_after_consuming_failing").await; + + assert_eq!( + offset, None, + "with offset_commit = after_consuming a rejected batch must leave the offset unstored so \ + the messages are redelivered, but the server reported {offset:?}" + ); +} + +#[iggy_harness( + server(connectors_runtime( + config_path = "tests/connectors/runtime/offset_commit_after_polling_failing.toml" + )), + seed = seeds::connector_stream +)] +async fn given_after_polling_when_sink_rejects_batch_should_still_advance_offset( + harness: &TestHarness, +) { + send_test_messages(harness).await; + + let last_offset = (MESSAGE_COUNT - 1) as u64; + let offset = + wait_for_stored_offset(harness, "offset_commit_after_polling_failing", last_offset).await; + + assert_eq!( + offset, + Some(last_offset), + "the default after_polling mode commits at poll time, so a rejected batch still advances \ + the offset and those messages are lost - this is the at-most-once behaviour that \ + after_consuming exists to avoid" + ); + + let status = wait_for_sink_status(harness, ConnectorStatus::Error).await; + assert_eq!( + status, + ConnectorStatus::Error, + "a rejection stops the sink in both modes - continuing would commit every later batch at \ + poll time and drain the topic into a target that is already refusing writes" + ); +} + +#[iggy_harness( + server(connectors_runtime( + config_path = "tests/connectors/runtime/offset_commit_multi_topic.toml" + )), + seed = seeds::connector_multi_topic_stream +)] +async fn given_multi_topic_sink_when_one_topic_rejects_should_stop_every_topic( + harness: &TestHarness, +) { + send_test_messages_to(harness, names::TOPIC).await; + + let status = wait_for_sink_status(harness, ConnectorStatus::Error).await; + assert_eq!( + status, + ConnectorStatus::Error, + "the rejecting topic should drive the connector to Error" + ); + + // The sink is configured to reject only TOPIC, so TOPIC_2 would be accepted + // and its offset committed if its task were still alive. Sending after the + // halt makes a committed offset here mean exactly one thing: the sibling + // outlived the failure. Both topics share one plugin instance and so one + // target, and the instance is the failure domain, not the task. + send_test_messages_to(harness, names::TOPIC_2).await; + sleep(NO_COMMIT_OBSERVATION_WINDOW).await; + + let offset = stored_offset_for(harness, "offset_commit_multi_topic", names::TOPIC_2).await; + assert_eq!( + offset, None, + "the sibling topic's task must stop with the instance, but it consumed a batch the sink \ + would have accepted and committed {offset:?}" + ); +} diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_consuming.toml b/core/integration/tests/connectors/runtime/offset_commit_after_consuming.toml new file mode 100644 index 0000000000..a0bdfe5eb0 --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_consuming.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "tests/connectors/runtime/offset_commit_after_consuming_config" diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_consuming_config/test_sink.toml b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_config/test_sink.toml new file mode 100644 index 0000000000..169035c607 --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_config/test_sink.toml @@ -0,0 +1,34 @@ +# 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. + +type = "sink" +key = "offset_commit_sink" +enabled = true +version = 0 +name = "Offset commit test sink" +path = "../../target/debug/libiggy_connector_test_sink" +offset_commit = "after_consuming" + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 10 +poll_interval = "5ms" +consumer_group = "offset_commit_after_consuming" + +[plugin_config] diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing.toml b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing.toml new file mode 100644 index 0000000000..2e7abd8da7 --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "tests/connectors/runtime/offset_commit_after_consuming_failing_config" diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing_config/test_sink.toml b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing_config/test_sink.toml new file mode 100644 index 0000000000..bc3fcd297d --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_consuming_failing_config/test_sink.toml @@ -0,0 +1,35 @@ +# 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. + +type = "sink" +key = "offset_commit_sink" +enabled = true +version = 0 +name = "Offset commit test sink" +path = "../../target/debug/libiggy_connector_test_sink" +offset_commit = "after_consuming" + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 10 +poll_interval = "5ms" +consumer_group = "offset_commit_after_consuming_failing" + +[plugin_config] +fail_after_batches = 0 diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing.toml b/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing.toml new file mode 100644 index 0000000000..d990f27e6b --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "tests/connectors/runtime/offset_commit_after_polling_failing_config" diff --git a/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing_config/test_sink.toml b/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing_config/test_sink.toml new file mode 100644 index 0000000000..ee6a6b057a --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_after_polling_failing_config/test_sink.toml @@ -0,0 +1,34 @@ +# 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. + +type = "sink" +key = "offset_commit_sink" +enabled = true +version = 0 +name = "Offset commit test sink" +path = "../../target/debug/libiggy_connector_test_sink" + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 10 +poll_interval = "5ms" +consumer_group = "offset_commit_after_polling_failing" + +[plugin_config] +fail_after_batches = 0 diff --git a/core/integration/tests/connectors/runtime/offset_commit_multi_topic.toml b/core/integration/tests/connectors/runtime/offset_commit_multi_topic.toml new file mode 100644 index 0000000000..0ff645c381 --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_multi_topic.toml @@ -0,0 +1,21 @@ +# 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. + + +[connectors] +config_type = "local" +config_dir = "tests/connectors/runtime/offset_commit_multi_topic_config" diff --git a/core/integration/tests/connectors/runtime/offset_commit_multi_topic_config/test_sink.toml b/core/integration/tests/connectors/runtime/offset_commit_multi_topic_config/test_sink.toml new file mode 100644 index 0000000000..b1a76434ee --- /dev/null +++ b/core/integration/tests/connectors/runtime/offset_commit_multi_topic_config/test_sink.toml @@ -0,0 +1,36 @@ +# 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. + +type = "sink" +key = "offset_commit_sink" +enabled = true +version = 0 +name = "Offset commit test sink" +path = "../../target/debug/libiggy_connector_test_sink" +offset_commit = "after_consuming" + +[[streams]] +stream = "test_stream" +topics = ["test_topic", "test_topic_2"] +schema = "json" +batch_length = 10 +poll_interval = "5ms" +consumer_group = "offset_commit_multi_topic" + +[plugin_config] +fail_after_batches = 0 +reject_topics = ["test_topic"]