hubcio commented on code in PR #3954:
URL: https://github.com/apache/iggy/pull/3954#discussion_r3924272682
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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));
Review Comment:
critical: this fires under the default `after_polling` too, and the SDK maps
every `Err` from a sink to `1`, so one transient write failure now permanently
stops the connector. nothing auto-restarts it - `restart_connector` is only
reachable over HTTP.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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))
Review Comment:
critical: any `IggyError` here, including a transient disconnect, kills the
connector with no retry. `RuntimeError::IggyError` prints just "Iggy error", so
`last_error` tells an operator nothing about this path.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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 {
Review Comment:
warning: a rejected batch takes the `Ok` arm, so the decode/prepare/ffi
histograms and the benchmark event record it as processed. contradicts the
comment right above about sub-stages only on success.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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
Review Comment:
warning: this commits the batch's last offset, marking everything up to it
consumed - including messages already dropped by decode or transform failures.
`after_consuming` is at-least-once per batch, not per message.
##########
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 {
Review Comment:
warning: `after_consuming` guarantees duplicates when the process dies
between the sink accepting a batch and `store_offset` landing. sinks have to be
idempotent - worth saying in the doc comment and the README.
##########
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<IggyMessage> = (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<u64> {
+ stored_offset_for(harness, consumer_group, names::TOPIC).await
+}
+
+async fn stored_offset_for(
+ harness: &TestHarness,
+ consumer_group: &str,
+ topic: &str,
+) -> Option<u64> {
+ 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<SinkInfoResponse> = 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<u64> {
+ 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;
Review Comment:
warning: a plugin that fails to load also gives `Error` status and no stored
offset, so this passes for the wrong reason. assert on `last_error.message`,
like `error_isolation.rs` does.
also at line 262.
##########
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<IggyMessage> = (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<u64> {
+ stored_offset_for(harness, consumer_group, names::TOPIC).await
+}
+
+async fn stored_offset_for(
+ harness: &TestHarness,
+ consumer_group: &str,
+ topic: &str,
+) -> Option<u64> {
+ 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<SinkInfoResponse> = 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<u64> {
+ 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;
Review Comment:
warning: nothing restarts the connector, so the redelivery this claims is
never observed. #3957 kills and restarts the node for the same assertion -
worth mirroring.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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(());
Review Comment:
critical: a sibling task woken by this breaks out and drops its half-filled
batch. under `after_polling` those offsets are already committed, so one topic
failing silently loses in-flight messages on the others.
##########
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;
Review Comment:
warning: `MESSAGE_COUNT` equals `batch_length`, so every test sends exactly
one batch. `AutoCommit::Disabled` with `PollingStrategy::next()` stalls once
the read position runs `batch_length` ahead of the stored offset, and nothing
here shows batch 2 arriving.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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);
Review Comment:
nit: `messages_processed` is skipped when `store_offset` fails, though the
sink did accept the batch. skipping it on rejection is right; on a commit
failure it is just wrong accounting.
##########
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<usize>,
+ reject_topics: Vec<String>,
+ state: Mutex<State>,
+}
+
+#[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<usize>,
+ /// 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<Vec<String>>,
+}
+
+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<ConsumedMessage>,
+ ) -> 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
Review Comment:
nit: `batches_consumed` is one counter across all topics but the threshold
is checked per topic. fine at `0`, meaningless at any other value.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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);
Review Comment:
nit: counted here and again when the `Err` reaches `spawn_consume_tasks`.
the failure path just above does not increment locally.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -737,7 +785,7 @@ async fn process_messages(
})?;
let ffi_start = Instant::now();
- (consume)(
+ let consume_result = (consume)(
Review Comment:
nit: nothing unit-tests `process_messages`. stub `extern "C"` callbacks
cover both the zero and non-zero status paths without needing a server.
##########
core/connectors/runtime/src/error.rs:
##########
@@ -27,6 +27,8 @@ pub enum RuntimeError {
FailedToSerializeMessagesMetadata,
#[error("Failed to serialize raw messages")]
Review Comment:
nit: positional tuple variant - named fields read better, and this message
is the only thing a test or an operator can match on.
##########
core/integration/tests/connectors/runtime/mod.rs:
##########
@@ -17,4 +17,5 @@
mod benchmark;
mod error_isolation;
+mod offset_commit;
Review Comment:
nit: rustfmt wants this after `mod http_state;` - this is the `cargo fmt`
failure.
##########
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.
+
+
Review Comment:
nit: extra blank line, drop one. this alone fails both taplo and the license
header check.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -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,
Review Comment:
simplification: the `Err` arm is unreachable - line 446 returns first. this
block also repeats the `match &result` directly below it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
