This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new bb1ef105e fix(connectors): propagate inner error messages in
RuntimeError (#4342)
bb1ef105e is described below
commit bb1ef105e54483aea94864c694457f42688cdfa7
Author: Daniil Novikov <[email protected]>
AuthorDate: Wed Sep 30 13:19:31 2026 +0200
fix(connectors): propagate inner error messages in RuntimeError (#4342)
---
core/connectors/runtime/src/error.rs | 50 +++++++++++++++++++---
.../tests/connectors/runtime/error_isolation.rs | 48 ++++++++++++++++++++-
.../connectors/runtime/sink_missing_stream.toml | 20 +++++++++
.../stdout_missing_stream.toml | 39 +++++++++++++++++
.../sink_missing_stream_config/stdout_valid.toml | 39 +++++++++++++++++
5 files changed, 189 insertions(+), 7 deletions(-)
diff --git a/core/connectors/runtime/src/error.rs
b/core/connectors/runtime/src/error.rs
index f0f5d11b0..156427b13 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -32,7 +32,7 @@ pub enum RuntimeError {
FailedToSerializeMessagesMetadata,
#[error("Failed to serialize raw messages")]
FailedToSerializeRawMessages,
- #[error("Connector SDK error")]
+ #[error(transparent)]
ConnectorSdkError(#[from] iggy_connector_sdk::Error),
/// A classified state-store failure while loading an enabled source's
/// state. Process-level: treating it as "no state" would silently rewind
@@ -43,16 +43,20 @@ pub enum RuntimeError {
connector_key: String,
source: iggy_connector_sdk::Error,
},
- #[error("Iggy client error")]
- IggyClient(#[from] iggy::prelude::ClientError),
- #[error("Iggy error")]
+ /// Passed through as-is. An `IggyError` that originated on the server
+ /// loses its fields on the wire (apache/iggy#3735): the client reads
+ /// only the status code and fills the rest with defaults, so ids in the
+ /// message can be zero placeholders and not real ids. This is not
+ /// specific to this variant. It applies wherever a server-originated
+ /// `IggyError` is wrapped or stringified, such as
+ /// `iggy::prelude::ClientError::SdkError` or a message built by a
+ /// connector plugin.
+ #[error(transparent)]
IggyError(#[from] iggy::prelude::IggyError),
#[error("Missing Iggy credentials")]
MissingIggyCredentials,
#[error("Missing TLS certificate file")]
MissingTlsCertificateFile,
- #[error("JSON error")]
- JsonError(#[from] serde_json::Error),
#[error("Sink not found with key: {0}")]
SinkNotFound(String),
#[error("Sink config not found with key: {0}, version: {1}")]
@@ -94,3 +98,37 @@ impl RuntimeError {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy::prelude::IggyError;
+
+ #[test]
+ fn given_iggy_error_when_displayed_should_include_inner_message() {
+ let inner = IggyError::StreamNameNotFound("orders".to_owned());
+ let expected = inner.to_string();
+
+ let error = RuntimeError::from(inner);
+
+ assert_eq!(
+ error.to_string(),
+ expected,
+ "RuntimeError should carry the IggyError message"
+ );
+ }
+
+ #[test]
+ fn given_connector_sdk_error_when_displayed_should_include_inner_message()
{
+ let inner = iggy_connector_sdk::Error::InitError("bad
credentials".to_owned());
+ let expected = inner.to_string();
+
+ let error = RuntimeError::from(inner);
+
+ assert_eq!(
+ error.to_string(),
+ expected,
+ "RuntimeError should carry the SDK error message"
+ );
+ }
+}
diff --git a/core/integration/tests/connectors/runtime/error_isolation.rs
b/core/integration/tests/connectors/runtime/error_isolation.rs
index a3a899b00..7c918abd4 100644
--- a/core/integration/tests/connectors/runtime/error_isolation.rs
+++ b/core/integration/tests/connectors/runtime/error_isolation.rs
@@ -28,7 +28,8 @@
//! `source::init` and `sink::init`:
//! * plugin path resolution failure (missing `.so`),
//! * source state-load failure (unreachable state file),
-//! * post-container setup failure (invalid duration in stream config).
+//! * post-container setup failure (invalid duration in stream config),
+//! * post-container setup failure surfaced as an Iggy error (missing
stream).
use iggy_common::{Identifier, IggyMessage, MessageClient, Partitioning};
use iggy_connector_sdk::api::{
@@ -398,3 +399,48 @@ async fn
given_sink_transform_error_when_batch_processed_should_not_count_filter
);
assert_eq!(sink.status, ConnectorStatus::Running);
}
+
+#[iggy_harness(
+ server(connectors_runtime(
+ config_path = "tests/connectors/runtime/sink_missing_stream.toml"
+ )),
+ seed = seeds::connector_stream
+)]
+async fn
given_sink_with_missing_stream_when_runtime_starts_should_expose_iggy_error_reason(
+ harness: &TestHarness,
+) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let http_client = Client::new();
+
+ assert_runtime_healthy(&http_client, &api_address).await;
+ let sinks = fetch_sinks(&http_client, &api_address).await;
+
+ let missing_stream_sink = sinks
+ .iter()
+ .find(|sink| sink.key == "stdout_missing_stream")
+ .expect("Sink with a missing stream should be reported");
+ assert_eq!(missing_stream_sink.status, ConnectorStatus::Error);
+ let last_error = missing_stream_sink
+ .last_error
+ .as_ref()
+ .expect("Sink with a missing stream should expose a last_error");
+ assert!(
+ last_error.message.contains("no_such_stream")
+ && last_error.message.contains("was not found"),
+ "last_error should carry the SDK's StreamNameNotFound message, got:
{}",
+ last_error.message
+ );
+
+ let valid_sink = sinks
+ .iter()
+ .find(|sink| sink.key == "stdout_valid")
+ .expect("Healthy sibling sink should be reported");
+ assert_eq!(valid_sink.status, ConnectorStatus::Running);
+ assert!(
+ valid_sink.last_error.is_none(),
+ "Healthy sibling sink should have no last_error"
+ );
+}
diff --git a/core/integration/tests/connectors/runtime/sink_missing_stream.toml
b/core/integration/tests/connectors/runtime/sink_missing_stream.toml
new file mode 100644
index 000000000..842741014
--- /dev/null
+++ b/core/integration/tests/connectors/runtime/sink_missing_stream.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/sink_missing_stream_config"
diff --git
a/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_missing_stream.toml
b/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_missing_stream.toml
new file mode 100644
index 000000000..344b61859
--- /dev/null
+++
b/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_missing_stream.toml
@@ -0,0 +1,39 @@
+# 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.
+
+# Deliberately broken sink: the configured stream does not exist on the server.
+# Used to assert that the Iggy error reason reaches `last_error` instead of a
+# generic "Iggy error" label.
+
+type = "sink"
+key = "stdout_missing_stream"
+enabled = true
+version = 0
+name = "Stdout missing stream sink"
+path = "../../target/debug/libiggy_connector_stdout_sink"
+verbose = true
+
+[[streams]]
+stream = "no_such_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "stdout_missing_stream_sink_connector"
+
+[plugin_config]
+print_payload = true
diff --git
a/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_valid.toml
b/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_valid.toml
new file mode 100644
index 000000000..0670d68fd
--- /dev/null
+++
b/core/integration/tests/connectors/runtime/sink_missing_stream_config/stdout_valid.toml
@@ -0,0 +1,39 @@
+# 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.
+
+# Healthy companion sink: verifies that a sibling connector with valid
+# configuration still starts and runs when another connector in the same
+# runtime fails initialization.
+
+type = "sink"
+key = "stdout_valid"
+enabled = true
+version = 0
+name = "Stdout valid sink"
+path = "../../target/debug/libiggy_connector_stdout_sink"
+verbose = true
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "stdout_valid_sink_connector"
+
+[plugin_config]
+print_payload = true
