You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/20 17:48:13 UTC
(iggy) branch master updated: fix(connectors): exclude key and version from connector env mappings (#4486)
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 d9c8a50d2 fix(connectors): exclude key and version from connector env mappings (#4486)
d9c8a50d2 is described below
commit d9c8a50d21b572b02f1f4ddf57dfcaf2950eb6c3
Author: Mit Shah <84...@users.noreply.github.com>
AuthorDate: Tue Oct 20 23:18:09 2026 +0530
fix(connectors): exclude key and version from connector env mappings (#4486)
---
.../configs/src/configs_impl/typed_env_provider.rs | 33 +++++++++-
core/connectors/runtime/src/configs/connectors.rs | 26 ++++++++
.../src/configs/connectors/local_provider.rs | 74 ++++++++++++++++++++++
core/connectors/runtime/src/main.rs | 1 -
4 files changed, 132 insertions(+), 2 deletions(-)
diff --git a/core/configs/src/configs_impl/typed_env_provider.rs b/core/configs/src/configs_impl/typed_env_provider.rs
index 47bef65c6..f802414fc 100644
--- a/core/configs/src/configs_impl/typed_env_provider.rs
+++ b/core/configs/src/configs_impl/typed_env_provider.rs
@@ -294,7 +294,16 @@ impl<T: ConfigEnvMappings> TypedEnvProvider<T> {
}
WarningContext::ConnectorConfig(prefix) => {
let plugin_config_prefix = format!("{}{PLUGIN_CONFIG_ENV_SEGMENT}", prefix);
- key.starts_with(&plugin_config_prefix)
+ if key.starts_with(&plugin_config_prefix) {
+ true
+ } else if key == format!("{prefix}KEY") || key == format!("{prefix}VERSION") {
+ info!(
+ "Environment variable '{key}' is ignored: connector identity comes from the config file."
+ );
+ true
+ } else {
+ false
+ }
}
};
if should_skip {
@@ -653,4 +662,26 @@ mod tests {
unsafe { env::remove_var(name) };
}
}
+
+ /// Retired KEY/VERSION overrides must not trip the unknown-variable scan.
+ #[test]
+ #[serial_test::serial]
+ fn retired_connector_identity_vars_are_skipped_by_the_unknown_variable_scan() {
+ let prefix = "IGGY_CONNECTORS_SINK_TEST_";
+ let retired = [format!("{prefix}KEY"), format!("{prefix}VERSION")];
+ for name in &retired {
+ // SAFETY: the race is process-wide, not per key: `set_var` is unsound
+ // against any concurrent environment access. `serial_test::serial` on
+ // this test is what prevents that.
+ unsafe { env::set_var(name, "stale") };
+ }
+
+ TypedEnvProvider::<TestConfig>::new(prefix, &[])
+ .warn_unknown_env_vars_inner(WarningContext::ConnectorConfig(prefix));
+
+ for name in &retired {
+ // SAFETY: paired with the set above.
+ unsafe { env::remove_var(name) };
+ }
+ }
}
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index 278481f4c..2a51f64f6 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -176,8 +176,11 @@ impl CreateSinkConfig {
#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct SinkConfig {
+ // Identity comes from the file: an env KEY would rename the sink and move its consumer group.
+ #[config_env(skip)]
pub key: String,
pub enabled: bool,
+ #[config_env(skip)]
pub version: u64,
pub name: String,
pub path: String,
@@ -229,8 +232,11 @@ impl CreateSourceConfig {
#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct SourceConfig {
+ // Identity comes from the file: an env KEY would orphan saved state.
+ #[config_env(skip)]
pub key: String,
pub enabled: bool,
+ #[config_env(skip)]
pub version: u64,
pub name: String,
pub path: String,
@@ -488,6 +494,26 @@ impl ConnectorsConfig {
#[cfg(test)]
mod tests {
use super::*;
+ use configs::ConfigEnvMappings;
+
+ #[test]
+ fn given_sink_and_source_configs_when_mapped_should_exclude_key_and_version() {
+ for (name, mappings) in [
+ ("sink", SinkConfig::env_mappings()),
+ ("source", SourceConfig::env_mappings()),
+ ] {
+ assert!(
+ mappings.iter().any(|mapping| mapping.env_name == "ENABLED"),
+ "{name} should map ENABLED from the environment"
+ );
+ assert!(
+ mappings
+ .iter()
+ .all(|mapping| !matches!(mapping.env_name, "KEY" | "VERSION")),
+ "{name} maps key or version from the environment"
+ );
+ }
+ }
#[test]
fn given_single_component_key_when_parsed_should_succeed() {
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index 272e5fc8d..9551c49d3 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -229,6 +229,12 @@ impl LocalConnectorsConfigProvider<Created> {
});
let connector_id: ConnectorId = (&connector_config).into();
let version = connector_config.version();
+ Self::reject_conflicting_identity_overrides(
+ &base_config,
+ version,
+ &path,
+ |name| std::env::var(name).ok(),
+ )?;
match connector_config {
ConnectorConfig::Sink(mut sink_config) => {
@@ -342,6 +348,37 @@ impl<S: ProviderState> LocalConnectorsConfigProvider<S> {
})
}
+ /// `KEY` and `VERSION` no longer come from the environment. An override that
+ /// differs from the file would silently select another checkpoint (sources)
+ /// or consumer group (sinks) on upgrade, so fail instead of ignoring it.
+ fn reject_conflicting_identity_overrides(
+ base_config: &BaseConnectorConfig,
+ version: u64,
+ path: &str,
+ env: impl Fn(&str) -> Option<String>,
+ ) -> Result<(), RuntimeError> {
+ let key = base_config.key();
+ let prefix = connector_env_prefix(
+ &base_config.connector_type().to_uppercase(),
+ &key.to_uppercase(),
+ );
+ for (suffix, file_value) in [("KEY", key.to_owned()), ("VERSION", version.to_string())] {
+ let name = format!("{prefix}{suffix}");
+ if let Some(value) = env(&name)
+ && value != file_value
+ {
+ return Err(RuntimeError::InvalidConfiguration(format!(
+ "Environment variable '{name}' is set to '{value}' but '{path}' defines {} '{file_value}'. \
+ Connector key and version can no longer be overridden from the environment: \
+ remove '{name}', or put the value in the config file (a new key or version \
+ starts without the previous state).",
+ suffix.to_lowercase()
+ )));
+ }
+ }
+ Ok(())
+ }
+
fn apply_plugin_config_env_overrides(
plugin_config: &mut Option<serde_json::Value>,
base_config: &BaseConnectorConfig,
@@ -908,6 +945,43 @@ mod tests {
assert_eq!(entries, vec!["source_random_0.toml"]);
}
+ #[test]
+ fn given_identity_env_overrides_when_checked_should_reject_only_conflicts() {
+ let base_config = BaseConnectorConfig::Source {
+ key: "stdout".to_owned(),
+ };
+ let check = |env: &[(&str, &str)]| {
+ LocalConnectorsConfigProvider::<Initialized>::reject_conflicting_identity_overrides(
+ &base_config,
+ 2,
+ "source_stdout_2.toml",
+ |name| {
+ env.iter()
+ .find(|(env_name, _)| *env_name == name)
+ .map(|(_, value)| (*value).to_owned())
+ },
+ )
+ };
+
+ assert!(check(&[]).is_ok());
+ assert!(
+ check(&[
+ ("IGGY_CONNECTORS_SOURCE_STDOUT_KEY", "stdout"),
+ ("IGGY_CONNECTORS_SOURCE_STDOUT_VERSION", "2"),
+ ])
+ .is_ok()
+ );
+ let key_error = check(&[("IGGY_CONNECTORS_SOURCE_STDOUT_KEY", "other")])
+ .unwrap_err()
+ .to_string();
+ assert!(
+ key_error.contains("IGGY_CONNECTORS_SOURCE_STDOUT_KEY"),
+ "{key_error}"
+ );
+ assert!(key_error.contains("remove"), "{key_error}");
+ assert!(check(&[("IGGY_CONNECTORS_SOURCE_STDOUT_VERSION", "3")]).is_err());
+ }
+
#[test]
fn given_runtime_format_env_when_loading_configs_should_preserve_plugin_overrides() {
const CHILD_PROCESS_ENV: &str = "IGGY_CONNECTORS_FORMAT_OVERRIDE_TEST";
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index 87245472c..a2cb46b91 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -186,7 +186,6 @@ fn print_config_env_vars() -> std::io::Result<()> {
let plugin_config = format!("{prefix}{PLUGIN_CONFIG_ENV_SEGMENT}<FIELD>");
templates
.iter()
- .filter(|template| !matches!(template.env_name, "KEY" | "VERSION"))
.map(move |template| format!("{prefix}{}", template.env_name))
.chain(std::iter::once(plugin_config))
});