From 91d7da502b849a3ee4e9c52b278f7b5dc368d081 Mon Sep 17 00:00:00 2001 From: minleejae <55116920+minleejae@users.noreply.github.com> Date: Sat, 22 Aug 2026 11:04:43 +0900 Subject: [PATCH 1/2] fix(connectors): propagate sink plugin consume failure status --- core/connectors/runtime/src/error.rs | 2 + core/connectors/runtime/src/sink.rs | 123 ++++++++++++++++++++++++++- 2 files changed, 124 insertions(+), 1 deletion(-) diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs index a8d0ba7647..893d108da2 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 plugin consume failed with status: {status} (plugin ID: {plugin_id})")] + SinkConsumeFailed { plugin_id: u32, status: i32 }, #[error("Connector SDK error")] ConnectorSdkError(#[from] iggy_connector_sdk::Error), #[error("Iggy client error")] diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs index 7a17724510..0b54f3bb68 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs @@ -737,7 +737,7 @@ async fn process_messages( })?; let ffi_start = Instant::now(); - (consume)( + let status = (consume)( plugin_id, topic_meta.as_ptr(), topic_meta.len(), @@ -748,6 +748,18 @@ async fn process_messages( ); let ffi_elapsed = ffi_start.elapsed(); + // The status code is the plugin's only channel for reporting a failed write: + // the SDK returns non-zero when the sink's consume() errors or the batch cannot + // be deserialized. Ignoring it would count the batch as processed and advance + // consumer offsets over messages the sink never stored — the same silent-loss + // class that the iggy_sink_open status check prevents at startup. + if status != 0 { + error!( + "Sink plugin consume failed with status: {status} for sink connector with ID: {plugin_id}" + ); + return Err(RuntimeError::SinkConsumeFailed { plugin_id, status }); + } + Ok(SinkBatchTiming { processed_count, decode_elapsed, @@ -760,3 +772,112 @@ struct SinkBatchTiming { decode_elapsed: Duration, ffi_elapsed: Duration, } + +#[cfg(test)] +mod tests { + use super::*; + use iggy_connector_sdk::decoders::json::JsonStreamDecoder; + + extern "C" fn consume_ok( + _id: u32, + _topic_meta_ptr: *const u8, + _topic_meta_len: usize, + _messages_meta_ptr: *const u8, + _messages_meta_len: usize, + _messages_ptr: *const u8, + _messages_len: usize, + ) -> i32 { + 0 + } + + extern "C" fn consume_fail( + _id: u32, + _topic_meta_ptr: *const u8, + _topic_meta_len: usize, + _messages_meta_ptr: *const u8, + _messages_meta_len: usize, + _messages_ptr: *const u8, + _messages_len: usize, + ) -> i32 { + 1 + } + + fn fixtures() -> ( + TopicMetadata, + MessagesMetadata, + Vec, + Arc, + Arc, + SinkLabels, + ) { + let topic_metadata = TopicMetadata { + stream: "test-stream".to_owned(), + topic: "test-topic".to_owned(), + }; + let messages_metadata = MessagesMetadata { + partition_id: 1, + current_offset: 0, + schema: Schema::Json, + }; + let messages = vec![ + IggyMessage::builder() + .payload(br#"{"key":"value"}"#.to_vec().into()) + .build() + .expect("test message should be valid"), + ]; + ( + topic_metadata, + messages_metadata, + messages, + Arc::new(JsonStreamDecoder), + Arc::new(Metrics::init()), + SinkLabels::new("test-sink"), + ) + } + + #[tokio::test] + async fn given_zero_status_when_plugin_consumes_should_return_timing() { + let (topic_metadata, messages_metadata, messages, decoder, metrics, labels) = fixtures(); + let result = process_messages( + 1, + messages_metadata, + &topic_metadata, + messages, + &(consume_ok as ConsumeCallback), + &Vec::new(), + &decoder, + &metrics, + &labels, + ) + .await; + match result { + Ok(timing) => assert_eq!(timing.processed_count, 1), + Err(error) => panic!("expected success, got error: {error}"), + } + } + + #[tokio::test] + async fn given_nonzero_status_when_plugin_consume_fails_should_propagate_error() { + let (topic_metadata, messages_metadata, messages, decoder, metrics, labels) = fixtures(); + let result = process_messages( + 2, + messages_metadata, + &topic_metadata, + messages, + &(consume_fail as ConsumeCallback), + &Vec::new(), + &decoder, + &metrics, + &labels, + ) + .await; + match result { + Err(RuntimeError::SinkConsumeFailed { plugin_id, status }) => { + assert_eq!(plugin_id, 2); + assert_eq!(status, 1); + } + Err(error) => panic!("expected SinkConsumeFailed, got: {error}"), + Ok(_) => panic!("expected SinkConsumeFailed, got success"), + } + } +} From 2334c91e53b7a4ef76854ebf6b0c6edc6bbee6b2 Mon Sep 17 00:00:00 2001 From: minleejae <55116920+minleejae@users.noreply.github.com> Date: Sun, 23 Aug 2026 14:11:28 +0900 Subject: [PATCH 2/2] fix(connectors): scope consume status comment to fail-fast semantics Signed-off-by: minleejae <55116920+minleejae@users.noreply.github.com> --- core/connectors/runtime/src/sink.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs index 0b54f3bb68..84609d0e70 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs @@ -748,11 +748,9 @@ async fn process_messages( ); let ffi_elapsed = ffi_start.elapsed(); - // The status code is the plugin's only channel for reporting a failed write: - // the SDK returns non-zero when the sink's consume() errors or the batch cannot - // be deserialized. Ignoring it would count the batch as processed and advance - // consumer offsets over messages the sink never stored — the same silent-loss - // class that the iggy_sink_open status check prevents at startup. + // Non-zero status is the plugin's only channel for reporting a failed write; propagate + // it so the runtime fails fast instead of counting the batch as processed. This does + // not redeliver the batch (offsets are already committed at poll time) — see #2927. if status != 0 { error!( "Sink plugin consume failed with status: {status} for sink connector with ID: {plugin_id}"