You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "gomnitrix (via GitHub)" <gi...@apache.org> on 2026/03/29 01:01:49 UTC
[PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
gomnitrix opened a new pull request, #3045:
URL: https://github.com/apache/iggy/pull/3045
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #2911
## Rationale
<!--
Why is this change needed? If the issue explains it well, a one-liner is fine.
-->
Production environments prefer default credential provider chain over static keys.
## What changed?
<!--
2-4 sentences. Problem first (before), then solution (after).
GOOD:
"Messages were unavailable when background message_saver committed the
journal and started async disk I/O before completion. Polling during
this window found neither journal nor disk data.
The fix freezes journal batches in the in-flight buffer before async persist."
GOOD:
"When many small messages accumulate in the journal, the flush passes
thousands of IO vectors to writev(), exceeding IOV_MAX (1024 on Linux)."
BAD:
- Walls of text
- "This PR adds..." (we can see the diff)
-->
The Iceberg Sink connector previously failed to initialize if `store_access_key_id` and `store_secret_access_key` were missing from the config, blocking the use of standard identity-based auth. This fix makes these config fields optional, allowing the underlying OpenDAL seamlessly fall back to the default credential provider chain. An isolated integration test was also added to verify this fallback behavior using environment variables.
## Local Execution
- Passed
- Pre-commit hooks ran
<!--
You must run your code locally before submitting.
"Relying on CI" is not acceptable - PRs from authors who haven't run the code will be closed.
Did you have `prek` installed? It runs automatically on commit and covers all project languages. See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md).
-->
## AI Usage
1. Which tools? Gemini 3.1 Pro
2. Scope of usage? Paired to implement the `iceberg_sink_uses_default_credential_chain` integration test.
3. How did you verify the generated code works correctly? Verified the test logic and ensured it correctly polls Iceberg snapshots without explicit static access key config.
4. Can you explain every line of the code if asked? yes
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3044526771
##########
core/integration/tests/connectors/fixtures/iceberg/container.rs:
##########
@@ -532,3 +535,39 @@ impl TestFixture for IcebergPreCreatedFixture {
self.inner.connectors_runtime_envs()
}
}
+
+pub struct IcebergEnvAuthFixture {
+ inner: IcebergPreCreatedFixture,
+}
+
+impl IcebergOps for IcebergEnvAuthFixture {
+ fn catalog_url(&self) -> &str {
+ self.inner.catalog_url()
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ self.inner.http_client()
+ }
+}
+
+#[async_trait]
+impl TestFixture for IcebergEnvAuthFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let inner = IcebergPreCreatedFixture::setup().await?;
+ // Set credentials before test server initialization.
+ unsafe {
Review Comment:
`unsafe { std::env::set_var }` without cleanup or safety justification. the tokio runtime is already running at this point, so concurrent env reads from other threads are UB. no `Drop` impl means these vars leak into subsequent tests.
the fix is straightforward: add `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` to the `HashMap` returned by `connectors_runtime_envs()` instead - the harness already passes that map to the child process via `Command::envs()` at `connectors_runtime.rs:157`, eliminating the `unsafe` entirely.
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
`get_props_s3()` handles each credential independently with separate `if let Some(...)` blocks. the partial-credential invariant (both-or-neither) is only enforced in `sink.rs::open()`, but `init_props` is `pub` - a future caller could bypass validation and get a silently half-configured props map. consider making `init_props` `pub(crate)` or adding both-or-neither validation here as defense-in-depth.
##########
core/integration/tests/connectors/iceberg/iceberg_sink.rs:
##########
@@ -207,3 +207,50 @@ async fn iceberg_sink_handles_bulk_messages(
assert_eq!(sinks.len(), 1);
assert!(sinks[0].last_error.is_none());
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/iceberg/sink_default_credentials.toml")),
+ seed = seeds::connector_stream
+)]
+async fn iceberg_sink_uses_default_credential_chain(
+ harness: &TestHarness,
+ fixture: IcebergEnvAuthFixture,
+) {
+ let client = harness.root_client().await.unwrap();
Review Comment:
bare `.unwrap()` calls here and at lines 220, 221, 227, 232 - the existing tests in this file consistently use `.expect("...")` with context messages.
##########
core/connectors/sinks/iceberg_sink/src/sink.rs:
##########
@@ -30,22 +30,32 @@ use tracing::{debug, error, info};
#[async_trait]
impl Sink for IcebergSink {
async fn open(&mut self) -> Result<(), Error> {
- let redacted_store_key = self
- .config
- .store_access_key_id
- .chars()
- .take(3)
- .collect::<String>();
- let redacted_store_secret = self
- .config
- .store_secret_access_key
- .chars()
- .take(3)
- .collect::<String>();
- info!(
- "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
- self.id, self.config.uri
- );
+ match (
+ &self.config.store_access_key_id,
+ &self.config.store_secret_access_key,
+ ) {
+ (Some(store_access_key_id), Some(store_secret_access_key)) => {
+ let redacted_store_key = store_access_key_id.chars().take(3).collect::<String>();
+ let redacted_store_secret =
+ store_secret_access_key.chars().take(3).collect::<String>();
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
+ self.id, self.config.uri
+ );
+ }
+ (None, None) => {
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}. No explicit credentials provided, falling back to default credential provider chain",
+ self.id, self.config.uri
+ );
+ }
+ _ => {
+ error!(
+ "Partially configured Iceberg credentials. You must provide both store_access_key_id and store_secret_access_key, or omit both."
+ );
+ return Err(Error::InvalidConfig);
Review Comment:
`Error::InvalidConfig` produces a bare "Invalid config" message. `Error::InvalidConfigValue(String)` exists (used by influxdb connectors) and carries a description - use it so the error is self-explanatory without needing to correlate with the `error!()` log above.
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
+ props.insert("s3.access-key-id".to_string(), access_key_id.clone());
+ }
+ if let Some(secret_access_key) = &config.store_secret_access_key {
+ props.insert(
+ "s3.secret-access-key".to_string(),
+ secret_access_key.clone(),
+ );
+ }
props.insert("s3.endpoint".to_string(), config.store_url.clone());
Ok(props)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::{IcebergSinkConfig, IcebergSinkStoreClass, IcebergSinkTypes};
+
+ #[test]
+ fn test_get_props_s3() {
Review Comment:
test covers both-`None` and both-`Some` but not the mixed case (one `Some`, one `None`). if validation moves into `get_props_s3` per the above comment, this test should verify the error path too.
##########
core/integration/tests/connectors/iceberg/default_credentials_config/config.toml:
##########
@@ -0,0 +1,45 @@
+# 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 = "iceberg"
+enabled = true
+version = 0
+name = "Iceberg sink"
+path = "../../target/release/libiggy_connector_iceberg_sink"
Review Comment:
`path` points to `target/release` but `ENV_SINK_PATH` always overrides to debug at runtime. not broken, but misleading. this is the same pattern as the original `config.toml` so not introduced by this PR.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3012929390
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
Did you manage to test end-to-end if the rest catalog falls back to the default credential manager if the access-key and secret are not present? If so please add the results in your PR. You can use this docker-compose to have a basic minio/spark-iceberg deployment locally that you can test on.
```yaml
services:
spark-iceberg:
image: tabulario/spark-iceberg
container_name: spark-iceberg
build: spark/
networks:
iceberg_net:
depends_on:
- rest
- minio
volumes:
- ./warehouse:/home/iceberg/warehouse
- ./notebooks:/home/iceberg/notebooks/notebooks
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
ports:
- 8888:8888
- 8080:8080
- 10000:10000
- 10001:10001
rest:
image: apache/iceberg-rest-fixture
container_name: iceberg-rest
networks:
iceberg_net:
ports:
- 8181:8181
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
- CATALOG_WAREHOUSE=s3://warehouse/
- CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
- CATALOG_S3_ENDPOINT=http://minio:9000
minio:
image: minio/minio
container_name: minio
environment:
- MINIO_ROOT_USER=admin
- MINIO_ROOT_PASSWORD=password
- MINIO_DOMAIN=minio
networks:
iceberg_net:
aliases:
- warehouse.minio
ports:
- 9001:9001
- 9000:9000
command: ["server", "/data", "--console-address", ":9001"]
mc:
depends_on:
- minio
image: minio/mc
container_name: mc
networks:
iceberg_net:
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
entrypoint: |
/bin/sh -c "
until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
/usr/bin/mc rm -r --force minio/warehouse;
/usr/bin/mc mb minio/warehouse;
/usr/bin/mc policy set public minio/warehouse;
tail -f /dev/null
"
networks:
iceberg_net:
```
##########
core/connectors/sinks/iceberg_sink/src/sink.rs:
##########
@@ -30,22 +30,22 @@ use tracing::{debug, error, info};
#[async_trait]
impl Sink for IcebergSink {
async fn open(&mut self) -> Result<(), Error> {
- let redacted_store_key = self
- .config
- .store_access_key_id
- .chars()
- .take(3)
- .collect::<String>();
- let redacted_store_secret = self
- .config
- .store_secret_access_key
- .chars()
- .take(3)
- .collect::<String>();
- info!(
- "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
- self.id, self.config.uri
- );
+ if let (Some(store_access_key_id), Some(store_secret_access_key)) = (
+ &self.config.store_access_key_id,
Review Comment:
If these are optional, maybe It'll be a good to fail if only one of them is present so we can fail early.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047537271
##########
core/connectors/sinks/iceberg_sink/src/sink.rs:
##########
@@ -30,22 +30,32 @@ use tracing::{debug, error, info};
#[async_trait]
impl Sink for IcebergSink {
async fn open(&mut self) -> Result<(), Error> {
- let redacted_store_key = self
- .config
- .store_access_key_id
- .chars()
- .take(3)
- .collect::<String>();
- let redacted_store_secret = self
- .config
- .store_secret_access_key
- .chars()
- .take(3)
- .collect::<String>();
- info!(
- "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
- self.id, self.config.uri
- );
+ match (
+ &self.config.store_access_key_id,
+ &self.config.store_secret_access_key,
+ ) {
+ (Some(store_access_key_id), Some(store_secret_access_key)) => {
+ let redacted_store_key = store_access_key_id.chars().take(3).collect::<String>();
+ let redacted_store_secret =
+ store_secret_access_key.chars().take(3).collect::<String>();
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
+ self.id, self.config.uri
+ );
+ }
+ (None, None) => {
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}. No explicit credentials provided, falling back to default credential provider chain",
+ self.id, self.config.uri
+ );
+ }
+ _ => {
+ error!(
+ "Partially configured Iceberg credentials. You must provide both store_access_key_id and store_secret_access_key, or omit both."
+ );
+ return Err(Error::InvalidConfig);
Review Comment:
I've switched to `Error::InvalidConfigValue`, I think it's now much cleaner!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3037354085
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
LGTM!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3035541908
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
Thanks for the docker-compose snippet! Here is how I verified it:
1. Automated CI Test: In this PR, I added a new integration test (iceberg_sink_uses_default_credential_chain) specifically to cover this fallback scenario. It has passed in the latest CI run: `PASS [ 18.521s] ( 770/1037) integration::mod connectors::iceberg::iceberg_sink::iceberg_sink_uses_default_credential_chain`
2. Manual E2E Testing: Following your suggestion, I also set up the docker environment locally and performed a manual test. Passing the credentials purely via env vars triggered the fallback mechanism correctly, and the messages were routed successfully:
```
# 1. The connector initializes and successfully falls back to default credentials:
2026-04-04T11:28:28.680638Z INFO connector: connector_target="iggy_connector_iceberg_sink::sink" Opened Iceberg sink connector with ID: 1 for URL: http://rest:8181. No explicit credentials provided, falling back to default credential provider chain
2026-04-04T11:28:28.680661Z INFO connector: connector_target="iggy_connector_iceberg_sink::sink" Configuring Iceberg catalog with the following config:
-region: us-east-1
-url: http://minio:9000
-store class: S3
-catalog type: REST
2026-04-04T11:28:28.692411Z INFO connector: connector_target="iggy_connector_iceberg_sink::router::static_router" Static router found 1 tables on iceberg catalog from 1 tables declared
2026-04-04T11:28:28.692612Z INFO iggy_connectors::sink: Sink container with name: Iceberg sink (iceberg), initialized successfully with ID: 1.
# 2. Sending a test message via CLI to the running server:
$ cargo run --bin iggy -- -u iggy -p ydYlMP0qiY0x8xF99h1bzBdS6ot6NvLTC message send test_stream test_topic "{\"id\": 1, \"name\": \"hello_iggy\"}"
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.34s
Running `target/debug/iggy -u iggy -p ydYlMP0qiY0x8xF99h1bzBdS6ot6NvLTC message send test_stream test_topic '{"id": 1, "name": "hello_iggy"}'`
Executing send messages to topic with ID: test_topic and stream with ID: test_stream
Sent messages to topic with ID: test_topic and stream with ID: test_stream
# 3. The connector immediately processes and routes it to Iceberg:
2026-04-04T11:29:54.643126Z INFO connector: connector_target="iggy_connector_iceberg_sink::router::static_router" Routed 1 messages to iceberg table messages successfully
```
I've also added the result into the PR details, hope this looks good to you!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3035525623
##########
core/connectors/sinks/iceberg_sink/src/sink.rs:
##########
@@ -30,22 +30,22 @@ use tracing::{debug, error, info};
#[async_trait]
impl Sink for IcebergSink {
async fn open(&mut self) -> Result<(), Error> {
- let redacted_store_key = self
- .config
- .store_access_key_id
- .chars()
- .take(3)
- .collect::<String>();
- let redacted_store_secret = self
- .config
- .store_secret_access_key
- .chars()
- .take(3)
- .collect::<String>();
- info!(
- "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***",
- self.id, self.config.uri
- );
+ if let (Some(store_access_key_id), Some(store_secret_access_key)) = (
+ &self.config.store_access_key_id,
Review Comment:
Thanks for pointing this out. I've update it with an exhaustive `match`, It now logs a error message and immediately returns an `Error::InvalidConfig` to fail fast.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3045:
URL: https://github.com/apache/iggy/pull/3045#issuecomment-4149162379
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3045?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `77.77778%` with `2 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 60.86%. Comparing base ([`411a697`](https://app.codecov.io/gh/apache/iggy/commit/411a697eb184ebc33d7ce4b862f51db49ce46ea5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`6d6ba12`](https://app.codecov.io/gh/apache/iggy/commit/6d6ba129e3367a84cfc5841257742d318e86c9c7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3045?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/connectors/sinks/iceberg\_sink/src/props.rs](https://app.codecov.io/gh/apache/iggy/pull/3045?src=pr&el=tree&filepath=core%2Fconnectors%2Fsinks%2Ficeberg_sink%2Fsrc%2Fprops.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NpbmtzL2ljZWJlcmdfc2luay9zcmMvcHJvcHMucnM=) | 77.77% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3045 +/- ##
=============================================
- Coverage 71.74% 60.86% -10.89%
Complexity 943 943
=============================================
Files 1121 1121
Lines 93800 93803 +3
Branches 71124 71127 +3
=============================================
- Hits 67301 57090 -10211
- Misses 23863 34184 +10321
+ Partials 2636 2529 -107
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.81% <77.77%> (-14.55%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.30% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `67.64% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.43% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.31% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3045/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `38.68% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3045?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/connectors/sinks/iceberg\_sink/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3045?src=pr&el=tree&filepath=core%2Fconnectors%2Fsinks%2Ficeberg_sink%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NpbmtzL2ljZWJlcmdfc2luay9zcmMvbGliLnJz) | `100.00% <ø> (ø)` | |
| [core/connectors/sinks/iceberg\_sink/src/sink.rs](https://app.codecov.io/gh/apache/iggy/pull/3045?src=pr&el=tree&filepath=core%2Fconnectors%2Fsinks%2Ficeberg_sink%2Fsrc%2Fsink.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NpbmtzL2ljZWJlcmdfc2luay9zcmMvc2luay5ycw==) | `100.00% <ø> (ø)` | |
| [core/connectors/sinks/iceberg\_sink/src/props.rs](https://app.codecov.io/gh/apache/iggy/pull/3045?src=pr&el=tree&filepath=core%2Fconnectors%2Fsinks%2Ficeberg_sink%2Fsrc%2Fprops.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NpbmtzL2ljZWJlcmdfc2luay9zcmMvcHJvcHMucnM=) | `80.00% <77.77%> (-8.24%)` | :arrow_down: |
... and [274 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3045/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047496093
##########
core/integration/tests/connectors/iceberg/iceberg_sink.rs:
##########
@@ -207,3 +207,50 @@ async fn iceberg_sink_handles_bulk_messages(
assert_eq!(sinks.len(), 1);
assert!(sinks[0].last_error.is_none());
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/iceberg/sink_default_credentials.toml")),
+ seed = seeds::connector_stream
+)]
+async fn iceberg_sink_uses_default_credential_chain(
+ harness: &TestHarness,
+ fixture: IcebergEnvAuthFixture,
+) {
+ let client = harness.root_client().await.unwrap();
Review Comment:
Fixed. I replaced all `unwrap()` calls with `expect()` to provide better error context.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3045:
URL: https://github.com/apache/iggy/pull/3045#issuecomment-4153196552
@EdgarModesto23 would you mind checking this?
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047569221
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
+ props.insert("s3.access-key-id".to_string(), access_key_id.clone());
+ }
+ if let Some(secret_access_key) = &config.store_secret_access_key {
+ props.insert(
+ "s3.secret-access-key".to_string(),
+ secret_access_key.clone(),
+ );
+ }
props.insert("s3.endpoint".to_string(), config.store_url.clone());
Ok(props)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::{IcebergSinkConfig, IcebergSinkStoreClass, IcebergSinkTypes};
+
+ #[test]
+ fn test_get_props_s3() {
Review Comment:
Updated the unit tests to cover the partial credential scenarios. Since adding the new cases made the test a bit long and bloated, I also split it up. Thanks!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on PR #3045:
URL: https://github.com/apache/iggy/pull/3045#issuecomment-4156419808
@hubcio sure!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047563734
##########
core/integration/tests/connectors/fixtures/iceberg/container.rs:
##########
@@ -532,3 +535,39 @@ impl TestFixture for IcebergPreCreatedFixture {
self.inner.connectors_runtime_envs()
}
}
+
+pub struct IcebergEnvAuthFixture {
+ inner: IcebergPreCreatedFixture,
+}
+
+impl IcebergOps for IcebergEnvAuthFixture {
+ fn catalog_url(&self) -> &str {
+ self.inner.catalog_url()
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ self.inner.http_client()
+ }
+}
+
+#[async_trait]
+impl TestFixture for IcebergEnvAuthFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let inner = IcebergPreCreatedFixture::setup().await?;
+ // Set credentials before test server initialization.
+ unsafe {
Review Comment:
Yes, this is a serious issue. I missed the concurrency risks while trying to figure out how the test environment handles variables. I'll spend more time digging into the implementation to avoid similar mistakes in the future. I've fixed it and thanks for the heads-up!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #3045:
URL: https://github.com/apache/iggy/pull/3045
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3037354085
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
LGTM! Great job
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3035541908
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
Thanks for the docker-compose snippet! Here is how I verified it:
1. Automated CI Test: In this PR, I added a new integration test (iceberg_sink_uses_default_credential_chain) specifically to cover this fallback scenario. It has passed in the latest CI run: `PASS [ 18.521s] ( 770/1037) integration::mod connectors::iceberg::iceberg_sink::iceberg_sink_uses_default_credential_chain`
2. Manual E2E Testing: Following your suggestion, I also set up the docker environment locally and performed a manual test. Passing the credentials purely via env vars triggered the fallback mechanism correctly, and the messages were routed successfully:
```
# 1. The connector initializes and successfully falls back to default credentials:
2026-04-04T11:28:28.680638Z INFO connector: connector_target="iggy_connector_iceberg_sink::sink" Opened Iceberg sink connector with ID: 1 for URL: http://rest:8181. No explicit credentials provided, falling back to default credential provider chain
2026-04-04T11:28:28.680661Z INFO connector: connector_target="iggy_connector_iceberg_sink::sink" Configuring Iceberg catalog with the following config:
-region: us-east-1
-url: http://minio:9000
-store class: S3
-catalog type: REST
2026-04-04T11:28:28.692411Z INFO connector: connector_target="iggy_connector_iceberg_sink::router::static_router" Static router found 1 tables on iceberg catalog from 1 tables declared
2026-04-04T11:28:28.692612Z INFO iggy_connectors::sink: Sink container with name: Iceberg sink (iceberg), initialized successfully with ID: 1.
# 2. Sending a test message via CLI to the running server:
$ cargo run --bin iggy -- -u iggy -p password message send test_stream test_topic "{\"id\": 1, \"name\": \"hello_iggy\"}"
Executing send messages to topic with ID: test_topic and stream with ID: test_stream
Sent messages to topic with ID: test_topic and stream with ID: test_stream
# 3. The connector immediately processes and routes it to Iceberg:
2026-04-04T11:29:54.643126Z INFO connector: connector_target="iggy_connector_iceberg_sink::router::static_router" Routed 1 messages to iceberg table messages successfully
```
I've also added the result into the PR details, hope this looks good to you!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047490878
##########
core/integration/tests/connectors/iceberg/default_credentials_config/config.toml:
##########
@@ -0,0 +1,45 @@
+# 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 = "iceberg"
+enabled = true
+version = 0
+name = "Iceberg sink"
+path = "../../target/release/libiggy_connector_iceberg_sink"
Review Comment:
I've updated the path to `debug` in this config to match the actual test environment and avoid confusion.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): support default credential provider chain for iceberg sink [iggy]
Posted by "gomnitrix (via GitHub)" <gi...@apache.org>.
gomnitrix commented on code in PR #3045:
URL: https://github.com/apache/iggy/pull/3045#discussion_r3047487858
##########
core/connectors/sinks/iceberg_sink/src/props.rs:
##########
@@ -30,14 +30,49 @@ pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>,
fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> {
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
- props.insert(
- "s3.access-key-id".to_string(),
- config.store_access_key_id.clone(),
- );
- props.insert(
- "s3.secret-access-key".to_string(),
- config.store_secret_access_key.clone(),
- );
+ if let Some(access_key_id) = &config.store_access_key_id {
Review Comment:
I've added the both-or-neither validation to `get_props_s3`. Thanks for pointing out!
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org