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/09/29 18:12:04 UTC
(iggy) branch master updated: feat(connectors): Delta sink: production feasibility improvements for S3 storage (#4219)
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 1f5d215b0 feat(connectors): Delta sink: production feasibility improvements for S3 storage (#4219)
1f5d215b0 is described below
commit 1f5d215b08269d3ff2b0efa72571d70ec53b89dc
Author: Daniil Novikov <36...@users.noreply.github.com>
AuthorDate: Tue Sep 29 20:11:59 2026 +0200
feat(connectors): Delta sink: production feasibility improvements for S3 storage (#4219)
---
core/connectors/sinks/delta_sink/README.md | 200 ++++++++++++++++-----
core/connectors/sinks/delta_sink/src/lib.rs | 1 +
core/connectors/sinks/delta_sink/src/sink.rs | 65 +++++--
core/connectors/sinks/delta_sink/src/storage.rs | 74 +++++---
.../tests/connectors/delta/delta_sink.rs | 118 +++++++++++-
.../delta/malformed_table_uri_config/config.toml | 38 ++++
.../connectors/delta/malformed_table_uri_sink.toml | 20 +++
.../delta/missing_secret_key_config/config.toml | 41 +++++
.../connectors/delta/missing_secret_key_sink.toml | 20 +++
.../tests/connectors/fixtures/delta/fixture.rs | 146 ++++++++++++++-
.../tests/connectors/fixtures/delta/mod.rs | 5 +-
.../integration/tests/connectors/fixtures/floci.rs | 1 +
core/integration/tests/connectors/fixtures/mod.rs | 5 +-
13 files changed, 642 insertions(+), 92 deletions(-)
diff --git a/core/connectors/sinks/delta_sink/README.md b/core/connectors/sinks/delta_sink/README.md
index 7a98642e4..95ffe9113 100644
--- a/core/connectors/sinks/delta_sink/README.md
+++ b/core/connectors/sinks/delta_sink/README.md
@@ -10,29 +10,152 @@ The Delta Lake Sink Connector allows you to consume messages from Iggy topics an
The table must already exist. The connector appends each successful nonempty batch in one Delta transaction and keeps its schema snapshot until restart. The plugin has no failed-batch retry loop; the Delta library can retry eligible commit conflicts and storage requests. Write or commit errors clear the writer buffers and return an error. The runtime uses consumer auto-commit and does not replay failed sink batches, so end-to-end at-least-once delivery is not guaranteed.
-## Configuration example
-
-### Local filesystem
-
-```toml
-[plugin_config]
-table_uri = "file:///tmp/iggy_delta_table"
+## How to configure a Delta Sink connector
+
+First, make sure that the Delta table already exists in the location you're providing. You can use this script for an example workload:
+
+```python
+import pyarrow as pa
+from deltalake import DeltaTable
+
+table_uri = "s3://my-bucket/tables/test"
+
+schema = pa.schema([
+ pa.field("user_id", pa.string(), nullable=True),
+ pa.field("user_type", pa.uint8(), nullable=True),
+ pa.field("email", pa.string(), nullable=True),
+ pa.field("source", pa.string(), nullable=True),
+ pa.field("state", pa.string(), nullable=True),
+ pa.field("message", pa.string(), nullable=True),
+ pa.field("created_at", pa.timestamp("us"), nullable=True),
+])
+
+DeltaTable.create(
+ table_uri,
+ schema,
+ name="test",
+ storage_options={"AWS_REGION": "us-east-1"},
+)
+
+print(f"Created table at {table_uri}")
```
-### AWS S3
-
-```toml
-[plugin_config]
-table_uri = "s3://my-bucket/delta-tables/users"
-storage_backend_type = "s3"
-aws_s3_access_key = "your-access-key"
-aws_s3_secret_key = "your-secret-key"
-aws_s3_region = "us-east-1"
-aws_s3_endpoint_url = "https://s3.amazonaws.com"
-aws_s3_allow_http = false
-```
+The configuration is usually written individually for every connector and consists of two parts: the runtime settings which are registering the sink and telling which streams should plug into it, and the plugin's settings themselves. Here's an example of a working configuration:
+
+ ```toml
+ type = "sink"
+ key = "delta"
+ enabled = true
+ version = 0
+ name = "Delta Lake sink"
+ path = "target/release/libiggy_connector_delta_sink" # make sure you have a compiled plugin binary in this path
+ verbose = true
+
+ # these settings are common between all sinks
+ [[streams]]
+ stream = "your_stream"
+ topics = ["topic_inside_of_your_stream"]
+ schema = "json"
+ # important: read about batch and poll interval settings below
+ batch_length = 10000
+ poll_interval = "3s"
+ consumer_group = "delta_sink_connector"
+
+ # these settings are specific to each plugin, and in case of Delta sink, to the type of storage used
+ [plugin_config]
+ # the table must exist in the given location
+ table_uri = "s3://my-bucket/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "us-east-1"
+ aws_s3_allow_http = false
+ ```
+
+### `[[streams]]` section
+
+- Find a topic that you need and the corresponding stream that this topic is in and add it into the configuration.
+- `batch_length` and `poll_interval` have to be carefully set for Delta tables. `poll_interval` sets the minimum gap between polls. Each poll returns immediately with whatever messages are currently available, up to `batch_length`. The Delta sink writes and commits exactly what one poll returns, so each Delta commit holds at most `batch_length` rows, and topic intake is capped at roughly `batch_length` messages per `poll_interval` per topic.
+ `batch_length` is a ceiling: it only produces large files when messages arrive fast enough between polls to fill it, so a low-throughput topic produces small, frequent commits regardless of how high `batch_length` is set. For Delta, writing means creating a log entry and a separate Parquet file, so a low `batch_length` (roughly < 1000) or a low-throughput topic results in lots of small files.
+ This is highly undesirable for the reader as it will have to read from many small files instead of a few larger ones. For query performance optimization, the system consuming these files will have to apply `OPTIMIZE` query in order to consolidate the files. We recommend setting `batch_length` pretty high based on your workload so that ideally the files are already optimized for reading.
+ The guideline recommended by the developers of Delta is to keep individual file sizes between 128 MB and 1 GB. In the context of this document, it means that ideally `batch_length` should be tuned so that each commit's file size falls within the suggested range, assuming your topic has enough throughput to fill it. The given range is only a rough guideline and you need to find a good setting based on the patterns of reading and writing in your systems.
+ A larger `batch_length` means a failed write loses more data: offsets commit at poll time, so if the Delta write or commit fails afterward, that whole batch is lost with no retry. It also sits in memory longer and in multiple copies — raw messages, coerced JSON, and the writer's internal buffer — until the commit succeeds.
+
+### Plugin configuration
+
+#### Attributes common to all types of storage
+
+- **table_uri** (required): Absolute URI to the Delta table. Supported schemes: `file://`, `s3://`, `az://`, `gs://`. Bare filesystem paths are not accepted; local tables must use the `file://` scheme.
+- **storage_backend_type** (optional): The cloud storage backend to use. One of `"s3"`, `"azure"`, or `"gcs"`. Omit for local filesystem tables.
-### Azure Blob Storage
+#### Local filesystem
+
+ ```toml
+ [plugin_config]
+ table_uri = "file:///tmp/iggy_delta_table"
+ ```
+
+#### AWS S3
+
+Currently the implementation offers two possible ways of accessing the bucket.
+
+1. No static keys — the AWS SDK discovers credentials on its own via its default credential chain (environment variables, shared config/profile including SSO, web identity federation, ECS/EKS container credentials, or EC2 instance metadata). This is the best practice recommended by AWS. For example, if running with an attached IAM role, the role assumed by the writer should allow these actions on the bucket, here is how a working policy looks in HCL:
+
+ If another Delta sink in the same `iggy-connectors` process is configured with static keys, opening it writes those keys into the shared process environment, and this keyless fallback can pick them up — which identity ends up writing depends on sink init order.
+
+ ```hcl
+ data "aws_iam_policy_document" "s3_write" {
+ statement {
+ sid = "BucketLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:ListBucket"
+ ]
+ resources = ["arn:aws:s3:::your-bucket-name"]
+ }
+
+ statement {
+ sid = "ObjectLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:PutObjectAcl",
+ "s3:AbortMultipartUpload"
+ ]
+ resources = ["arn:aws:s3:::your-bucket-name/*"]
+ }
+ }
+ ```
+
+ The corresponding plugin configuration will look like this:
+
+ ```toml
+ [plugin_config]
+ table_uri = "s3://my-bucket/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "us-east-1"
+ ```
+
+2. Long-term access key pairs (access + secret key), which AWS recommends to avoid for security reasons. The example configuration:
+
+ ```toml
+ [plugin_config]
+ table_uri = "s3://my-bucket/delta-tables/users"
+ storage_backend_type = "s3"
+ aws_s3_access_key = "your-access-key"
+ aws_s3_secret_key = "your-secret-key"
+ aws_s3_region = "us-east-1"
+ ```
+
+Parameter descriptions:
+
+- **aws_s3_access_key**: Optional. AWS access key ID. Can only be passed together with the secret key.
+- **aws_s3_secret_key**: Optional. AWS secret access key. Can only be passed together with the access key.
+- **aws_s3_region**: Required. AWS region (e.g. `us-east-1`).
+- **aws_s3_endpoint_url**: Optional. S3 endpoint URL. Use for S3-compatible services. Make sure that the URL implies the same region that is set in **aws_s3_region**, otherwise you'll have an error.
+ Setting a custom endpoint skips the AWS SDK's full credential chain: shared profile and SSO credentials stop working, and only static keys, web identity federation, container credentials, or instance metadata remain available. Set `AWS_FORCE_CREDENTIAL_LOAD=true` in the `iggy-connectors` process environment to restore the full chain.
+- **aws_s3_allow_http**: Optional. Set to `true` to allow HTTP connections (for local development).
+
+#### Azure Blob Storage
```toml
[plugin_config]
@@ -43,7 +166,14 @@ azure_storage_account_key = "account-key"
azure_container_name = "my-container"
```
-### Google Cloud Storage
+Required when `storage_backend_type = "azure"`.
+
+- **azure_storage_account_name**: Azure storage account name.
+- **azure_storage_account_key**: Azure storage account key. Provide exactly one of this or **azure_storage_sas_token**.
+- **azure_storage_sas_token**: Shared Access Signature token. Provide exactly one of this or **azure_storage_account_key**.
+- **azure_container_name**: Azure container name.
+
+#### Google Cloud Storage
```toml
[plugin_config]
@@ -52,34 +182,6 @@ storage_backend_type = "gcs"
gcs_service_account_key = '{"type": "service_account", "project_id": "...", ...}'
```
-## Configuration Options
-
-### Core
-
-- **table_uri** (required): Absolute URI to an existing Delta table. Use `file:///...` for a local path; bare filesystem paths are not accepted. Supported schemes: `file://`, `s3://`, `az://`, `gs://`.
-- **storage_backend_type** (optional): The cloud storage backend to use. One of `"s3"`, `"azure"`, or `"gcs"`. Omit for local filesystem tables.
-
-### AWS S3
-
-Required when `storage_backend_type = "s3"`.
-
-- **aws_s3_access_key**: AWS access key ID.
-- **aws_s3_secret_key**: AWS secret access key.
-- **aws_s3_region**: AWS region (e.g. `us-east-1`).
-- **aws_s3_endpoint_url** (optional): S3 endpoint URL. Useful for S3-compatible services like MinIO.
-- **aws_s3_allow_http** (optional, default `false`): Set to `true` to allow HTTP connections (for local development).
-
-### Azure Blob Storage
-
-When `storage_backend_type = "azure"`, provide the account name, container name, and exactly one of the account key or SAS token. Providing both authentication fields is an error.
-
-- **azure_storage_account_name**: Azure storage account name.
-- **azure_storage_account_key**: Azure storage account key.
-- **azure_storage_sas_token**: Shared Access Signature token.
-- **azure_container_name**: Azure container name.
-
-### Google Cloud Storage
-
Required when `storage_backend_type = "gcs"`.
- **gcs_service_account_key**: GCS service account JSON key (as a string). The bucket is inferred from the `gs://` URI in `table_uri`.
diff --git a/core/connectors/sinks/delta_sink/src/lib.rs b/core/connectors/sinks/delta_sink/src/lib.rs
index e7d7ad111..608e9b7fc 100644
--- a/core/connectors/sinks/delta_sink/src/lib.rs
+++ b/core/connectors/sinks/delta_sink/src/lib.rs
@@ -53,6 +53,7 @@ struct SinkState {
}
#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
pub struct DeltaSinkConfig {
pub table_uri: String,
#[serde(default)]
diff --git a/core/connectors/sinks/delta_sink/src/sink.rs b/core/connectors/sinks/delta_sink/src/sink.rs
index 4bba68853..36ffbe358 100644
--- a/core/connectors/sinks/delta_sink/src/sink.rs
+++ b/core/connectors/sinks/delta_sink/src/sink.rs
@@ -37,33 +37,64 @@ impl Sink for DeltaSink {
);
let table_url = url::Url::parse(&self.config.table_uri).map_err(|e| {
- error!("Failed to parse table URI '{}': {e}", self.config.table_uri);
- Error::InitError(format!("Invalid table URI: {e}"))
+ error!(
+ "Connector configuration: failed to parse table_uri = '{}': {e}.",
+ self.config.table_uri
+ );
+ Error::InvalidConfigValue(format!("table_uri: {e}"))
})?;
+ let table_uri = &self.config.table_uri;
info!("Parsed table URI: {}", table_url);
- let storage_options = build_storage_options(&self.config).map_err(|e| {
- error!("Invalid storage configuration: {e}");
- Error::InitError(format!("Invalid storage configuration: {e}"))
+ let storage_options = build_storage_options(&self.config).inspect_err(|e| {
+ error!("Connector configuration: invalid storage configuration. Error message: {e}");
})?;
- let table =
- match deltalake::open_table_with_storage_options(table_url, storage_options).await {
- Ok(table) => table,
- Err(e) => {
- error!("Failed to load Delta table: {e}");
- return Err(Error::InitError(format!("Failed to load Delta table: {e}")));
+ let builder = deltalake::DeltaTableBuilder::from_url(table_url)
+ .map_err(|e| {
+ error!("deltalake-rs interface: failed to configure with table_uri = '{table_uri}'. Check deltalake::DeltaTableBuilder::from_url docs and code to correct your table_uri. Error message: {e}");
+ Error::InvalidConfigValue(format!("table_uri = '{table_uri}' caused an error in deltalake-rs interface. Check deltalake::DeltaTableBuilder::from_url docs and code to correct your table_uri. Error message: {e}"))
+ })?
+ .with_storage_options(storage_options);
+ let mut table = builder.build().map_err(|e| {
+ error!("deltalake-rs interface: failed to configure with provided storage configuration. Error message: {e}");
+ Error::InitError(format!("deltalake-rs interface: failed to configure with provided storage configuration. Error message: {e}"))
+ })?;
+ let table_exists = table
+ .verify_deltatable_existence()
+ .await
+ .map_err(
+ |e| {
+ error!("deltalake-rs interface: failed to list table_uri '{table_uri}' directory to verify delta table existence. Make sure the destination exists and the access to the destination is set up correctly - read the Iggy delta connector docs for more information. Error message: {e}");
+ Error::InitError(format!("deltalake-rs interface: failed to list table_uri '{table_uri}' directory to verify delta table existence. Make sure the destination exists and the access to the destination is set up correctly - read the Iggy delta connector docs for more information. Error message: {e}"))
}
- };
+ )?;
+ if !table_exists {
+ error!(
+ "No delta table found in '{table_uri}'. Make sure to create the delta table in the destination manually or verify the validity of such table."
+ );
+ return Err(Error::InitError(format!(
+ "No delta table found in '{table_uri}'. Make sure to create the delta table in the destination manually or verify the validity of such table."
+ )));
+ }
+
+ table
+ .load()
+ .await
+ .map_err(|e| {
+ error!("deltalake-rs interface: failed to load the table's latest snapshot. See deltalake::DeltaTable::load for more information. Error message: {e}");
+ Error::InitError(format!("deltalake-rs interface: failed to load the table's latest snapshot. See deltalake::DeltaTable::load for more information. Error message: {e}"))
+ })?;
let kernel_schema = table
.snapshot()
.map_err(|e| {
- error!("Failed to get table snapshot: {e}");
- Error::InitError(format!("Failed to get table snapshot: {e}"))
+ error!("deltalake-rs interface: failed to get the table's latest snapshot. See deltalake::DeltaTable::snapshot for more information. Error message: {e}");
+ Error::InitError(format!("deltalake-rs interface: failed to get the table's latest snapshot. See deltalake::DeltaTable::snapshot for more information. Error message: {e}"))
})?
.schema();
+
// TODO: coercion tree is never refreshed if the schema changes concurrently,
// leading to opaque errors downstream.
let coercion_tree = create_coercion_tree(&kernel_schema);
@@ -107,9 +138,9 @@ impl Sink for DeltaSink {
return Ok(());
}
- // TODO: all partition consume() calls serialize on this single lock, holding it
- // through flush_and_commit() I/O. fix: per-partition writers keyed by partition_id.
- // Ref: https://github.com/apache/iggy/pull/2889/#discussion_r2936719763
+ // This lock is held across the write and flush_and_commit I/O below,
+ // serializing consume() for topics sharing this sink. Kept intentionally:
+ // see #3839 for why per-partition writers were researched and dropped.
let mut state_guard = self.state.lock().await;
let state = state_guard.as_mut().ok_or_else(|| {
error!("Delta sink state not initialized — was open() called?");
diff --git a/core/connectors/sinks/delta_sink/src/storage.rs b/core/connectors/sinks/delta_sink/src/storage.rs
index 3a0e919ba..b6dbcee0c 100644
--- a/core/connectors/sinks/delta_sink/src/storage.rs
+++ b/core/connectors/sinks/delta_sink/src/storage.rs
@@ -19,6 +19,7 @@ use crate::{DeltaSinkConfig, StorageBackendType};
use iggy_connector_sdk::Error;
use secrecy::ExposeSecret;
use std::collections::HashMap;
+use tracing::info;
pub(crate) fn build_storage_options(
config: &DeltaSinkConfig,
@@ -27,25 +28,42 @@ pub(crate) fn build_storage_options(
match config.storage_backend_type {
Some(StorageBackendType::S3) => {
- let access_key = config.aws_s3_access_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires 'aws_s3_access_key'".into())
- })?;
- let secret_key = config.aws_s3_secret_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires 'aws_s3_secret_key'".into())
+ match (
+ config.aws_s3_access_key.as_ref(),
+ config.aws_s3_secret_key.as_ref(),
+ ) {
+ (Some(access_key), Some(secret_key)) => {
+ info!("S3 backend: using static access key/secret key credentials");
+ opts.insert(
+ "AWS_ACCESS_KEY_ID".into(),
+ access_key.expose_secret().to_owned(),
+ );
+ opts.insert(
+ "AWS_SECRET_ACCESS_KEY".into(),
+ secret_key.expose_secret().to_owned(),
+ );
+ }
+ (None, None) => {
+ // Edge case: no key here means this sink relies on the AWS SDK's own
+ // credential discovery (e.g. IAM role). But if another Delta sink in
+ // this same process sets a key, opening it writes that key into the
+ // shared process environment, and this sink's fallback can silently
+ // read it too.
+ info!(
+ "S3 backend: no static keys provided, relying on AWS SDK's default credential chain"
+ );
+ }
+ _ => {
+ return Err(Error::InvalidConfigValue(
+ "aws_s3_access_key and aws_s3_secret_key must be provided together, or both omitted".into(),
+ ));
+ }
+ }
+
+ let region = config.aws_s3_region.as_ref().ok_or_else(|| {
+ Error::InvalidConfigValue("S3 backend requires 'aws_s3_region'".into())
})?;
- let region = config
- .aws_s3_region
- .as_ref()
- .ok_or_else(|| Error::InitError("S3 backend requires 'aws_s3_region'".into()))?;
- opts.insert(
- "AWS_ACCESS_KEY_ID".into(),
- access_key.expose_secret().to_owned(),
- );
- opts.insert(
- "AWS_SECRET_ACCESS_KEY".into(),
- secret_key.expose_secret().to_owned(),
- );
opts.insert("AWS_REGION".into(), region.clone());
if let Some(endpoint_url) = config.aws_s3_endpoint_url.as_ref() {
@@ -58,10 +76,12 @@ pub(crate) fn build_storage_options(
}
Some(StorageBackendType::Azure) => {
let account_name = config.azure_storage_account_name.as_ref().ok_or_else(|| {
- Error::InitError("Azure backend requires 'azure_storage_account_name'".into())
+ Error::InvalidConfigValue(
+ "Azure backend requires 'azure_storage_account_name'".into(),
+ )
})?;
let container_name = config.azure_container_name.as_ref().ok_or_else(|| {
- Error::InitError("Azure backend requires 'azure_container_name'".into())
+ Error::InvalidConfigValue("Azure backend requires 'azure_container_name'".into())
})?;
opts.insert("AZURE_STORAGE_ACCOUNT_NAME".into(), account_name.clone());
@@ -84,16 +104,16 @@ pub(crate) fn build_storage_options(
);
}
(Some(_), Some(_)) => {
- return Err(Error::InitError("Azure backend requires exactly one of 'azure_storage_account_key' or 'azure_storage_sas_token', but both were provided".into()));
+ return Err(Error::InvalidConfigValue("Azure backend requires exactly one of 'azure_storage_account_key' or 'azure_storage_sas_token', but both were provided".into()));
}
(None, None) => {
- return Err(Error::InitError("Azure backend requires one of 'azure_storage_account_key' or 'azure_storage_sas_token'".into()));
+ return Err(Error::InvalidConfigValue("Azure backend requires one of 'azure_storage_account_key' or 'azure_storage_sas_token'".into()));
}
}
}
Some(StorageBackendType::Gcs) => {
let service_account_key = config.gcs_service_account_key.as_ref().ok_or_else(|| {
- Error::InitError("GCS backend requires 'gcs_service_account_key'".into())
+ Error::InvalidConfigValue("GCS backend requires 'gcs_service_account_key'".into())
})?;
opts.insert(
@@ -206,6 +226,16 @@ mod tests {
assert!(build_storage_options(&config).is_err());
}
+ #[test]
+ fn s3_backend_without_access_and_secret_keys_succeeds() {
+ let mut config = s3_config();
+ config.aws_s3_access_key = None;
+ config.aws_s3_secret_key = None;
+ let opts = build_storage_options(&config).unwrap();
+ assert!(!opts.contains_key("AWS_ACCESS_KEY_ID"));
+ assert!(!opts.contains_key("AWS_SECRET_ACCESS_KEY"));
+ }
+
#[test]
fn s3_backend_missing_region_errors() {
let mut config = s3_config();
diff --git a/core/integration/tests/connectors/delta/delta_sink.rs b/core/integration/tests/connectors/delta/delta_sink.rs
index 71e6dd9bd..3ed8f82f7 100644
--- a/core/integration/tests/connectors/delta/delta_sink.rs
+++ b/core/integration/tests/connectors/delta/delta_sink.rs
@@ -16,7 +16,10 @@
// under the License.
use crate::connectors::create_test_messages;
-use crate::connectors::fixtures::{DeltaFixture, DeltaS3Fixture};
+use crate::connectors::fixtures::{
+ DeltaCorruptedLogFixture, DeltaFixture, DeltaS3Fixture, DeltaS3NoBucketFixture,
+ DeltaS3NoTableFixture, DeltaS3WrongCredentialsFixture,
+};
use bytes::Bytes;
use iggy::prelude::{IggyMessage, Partitioning};
use iggy_common::Identifier;
@@ -235,3 +238,116 @@ async fn delta_sink_writes_to_s3(harness: &TestHarness, fixture: DeltaS3Fixture)
assert_eq!(row_count, message_count);
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/delta/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_bucket_absence_on_s3(
+ harness: &TestHarness,
+ _fixture: DeltaS3NoBucketFixture,
+) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("failed to list table_uri") && logs.contains("ERROR"),
+ "the error is expected since the bucket is not created"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/delta/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_table_absence_on_s3(
+ harness: &TestHarness,
+ _fixture: DeltaS3NoTableFixture,
+) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("No delta table found in") && logs.contains("ERROR"),
+ "the error is expected since no table is written into the destination"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(
+ config_path = "tests/connectors/delta/malformed_table_uri_sink.toml"
+ )),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_malformed_table_uri(harness: &TestHarness) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("failed to parse table_uri") && logs.contains("ERROR"),
+ "the error is expected since the table_uri is malformed"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(
+ config_path = "tests/connectors/delta/missing_secret_key_sink.toml"
+ )),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_missing_secret_key_on_s3(harness: &TestHarness) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("Connector configuration: invalid storage configuration")
+ && logs.contains("ERROR"),
+ "the error is expected since only the access key is set and the secret key is missing"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/delta/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_invalid_credentials_on_s3(
+ harness: &TestHarness,
+ _fixture: DeltaS3WrongCredentialsFixture,
+) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("failed to list table_uri") && logs.contains("ERROR"),
+ "the error is expected since the S3 credentials are wrong"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/delta/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_corrupted_delta_log(
+ harness: &TestHarness,
+ _fixture: DeltaCorruptedLogFixture,
+) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ assert!(
+ logs.contains("failed to load the table's latest snapshot") && logs.contains("ERROR"),
+ "the error is expected since the delta log's commit file is corrupted"
+ );
+}
diff --git a/core/integration/tests/connectors/delta/malformed_table_uri_config/config.toml b/core/integration/tests/connectors/delta/malformed_table_uri_config/config.toml
new file mode 100644
index 000000000..89523244e
--- /dev/null
+++ b/core/integration/tests/connectors/delta/malformed_table_uri_config/config.toml
@@ -0,0 +1,38 @@
+# 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 malformed table_uri: no URL scheme, so `url::Url::parse`
+# rejects it before any storage backend is touched.
+
+type = "sink"
+key = "delta_malformed_uri"
+enabled = true
+version = 0
+name = "Delta malformed URI sink"
+path = "../../target/debug/libiggy_connector_delta_sink"
+verbose = true
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "delta_malformed_uri_sink_connector"
+
+[plugin_config]
+table_uri = "not-a-valid-table-uri"
diff --git a/core/integration/tests/connectors/delta/malformed_table_uri_sink.toml b/core/integration/tests/connectors/delta/malformed_table_uri_sink.toml
new file mode 100644
index 000000000..902b741dd
--- /dev/null
+++ b/core/integration/tests/connectors/delta/malformed_table_uri_sink.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/delta/malformed_table_uri_config"
diff --git a/core/integration/tests/connectors/delta/missing_secret_key_config/config.toml b/core/integration/tests/connectors/delta/missing_secret_key_config/config.toml
new file mode 100644
index 000000000..14f1bd46c
--- /dev/null
+++ b/core/integration/tests/connectors/delta/missing_secret_key_config/config.toml
@@ -0,0 +1,41 @@
+# 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 sets only the S3 access key: build_storage_options rejects
+# this before any storage backend is touched, so no MinIO/table is needed.
+
+type = "sink"
+key = "delta_missing_secret_key"
+enabled = true
+version = 0
+name = "Delta missing secret key sink"
+path = "../../target/debug/libiggy_connector_delta_sink"
+verbose = true
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "delta_missing_secret_key_sink_connector"
+
+[plugin_config]
+table_uri = "s3://iggy-sandbox/tables/test"
+storage_backend_type = "s3"
+aws_s3_access_key = "some-access-key"
+aws_s3_region = "us-east-1"
diff --git a/core/integration/tests/connectors/delta/missing_secret_key_sink.toml b/core/integration/tests/connectors/delta/missing_secret_key_sink.toml
new file mode 100644
index 000000000..b0963fb5e
--- /dev/null
+++ b/core/integration/tests/connectors/delta/missing_secret_key_sink.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/delta/missing_secret_key_config"
diff --git a/core/integration/tests/connectors/fixtures/delta/fixture.rs b/core/integration/tests/connectors/fixtures/delta/fixture.rs
index 9fa467091..6c0629c83 100644
--- a/core/integration/tests/connectors/fixtures/delta/fixture.rs
+++ b/core/integration/tests/connectors/fixtures/delta/fixture.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use std::{collections::HashMap, path::PathBuf};
+use std::{collections::HashMap, path::Path, path::PathBuf};
use async_trait::async_trait;
use deltalake::kernel::{DataType, PrimitiveType, StructField};
@@ -194,6 +194,68 @@ impl TestFixture for DeltaFixture {
}
}
+pub struct DeltaCorruptedLogFixture {
+ _temp_dir: TempDir,
+ table_path: PathBuf,
+}
+
+impl DeltaCorruptedLogFixture {
+ // Overwrites the version-0 commit file with garbage bytes but keeps its
+ // name intact: `is_delta_table_location` only lists filenames under
+ // `_delta_log`, so the table still "exists"; `DeltaTable::load` is what
+ // then fails parsing this file during log replay.
+ async fn corrupt_commit_log(table_path: &Path) -> Result<(), TestBinaryError> {
+ let commit_path = table_path
+ .join("_delta_log")
+ .join("00000000000000000000.json");
+ tokio::fs::write(&commit_path, b"not a valid delta log commit entry")
+ .await
+ .map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!(
+ "Failed to corrupt commit log at {}: {error}",
+ commit_path.display()
+ ),
+ })
+ }
+}
+
+#[async_trait]
+impl TestFixture for DeltaCorruptedLogFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let temp_dir = TempDir::new().map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!("Failed to create temp directory: {error}"),
+ })?;
+
+ let table_path = temp_dir.path().join("delta_table");
+ let table_uri = format!("file://{}", table_path.display());
+ DeltaFixture::create_table(&table_uri).await?;
+ Self::corrupt_commit_log(&table_path).await?;
+ info!(
+ "Delta corrupted-log fixture created with table path: {}",
+ table_path.display()
+ );
+
+ Ok(Self {
+ _temp_dir: temp_dir,
+ table_path,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let table_uri = format!("file://{}", self.table_path.display());
+
+ let mut envs = HashMap::new();
+ envs.insert(ENV_SINK_TABLE_URI.to_string(), table_uri);
+ envs.insert(
+ ENV_SINK_PATH.to_string(),
+ "../../target/debug/libiggy_connector_delta_sink".to_string(),
+ );
+ envs
+ }
+}
+
pub struct DeltaS3Fixture {
#[allow(dead_code)]
floci: FlociContainer,
@@ -300,3 +362,85 @@ impl TestFixture for DeltaS3Fixture {
envs
}
}
+
+pub struct DeltaS3NoTableFixture {
+ inner: DeltaS3Fixture,
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3NoTableFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let floci_name = fixtures::unique_container_name("floci-delta-no-table");
+
+ let floci = FlociContainer::start(None, &floci_name).await?;
+ let floci_endpoint = floci.endpoint.clone();
+ floci::create_bucket(&floci_endpoint, TEST_BUCKET).await?;
+
+ info!("Delta S3 'no table' fixture ready with Floci at {floci_endpoint}");
+
+ Ok(Self {
+ inner: DeltaS3Fixture {
+ floci,
+ floci_endpoint,
+ },
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
+
+pub struct DeltaS3NoBucketFixture {
+ inner: DeltaS3Fixture,
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3NoBucketFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let floci_name = fixtures::unique_container_name("floci-delta-no-bucket");
+
+ let floci = FlociContainer::start(None, &floci_name).await?;
+ let floci_endpoint = floci.endpoint.clone();
+
+ info!("Delta S3 'no bucket' fixture ready with Floci at {floci_endpoint}");
+
+ Ok(Self {
+ inner: DeltaS3Fixture {
+ floci,
+ floci_endpoint,
+ },
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
+
+pub struct DeltaS3WrongCredentialsFixture {
+ inner: DeltaS3Fixture,
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3WrongCredentialsFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ info!("Delta S3 'wrong credentials' fixture reusing a regular DeltaS3Fixture setup");
+ Ok(Self {
+ inner: DeltaS3Fixture::setup().await?,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.inner.connectors_runtime_envs();
+ envs.insert(
+ ENV_SINK_AWS_S3_ACCESS_KEY.to_string(),
+ "wrong-access-key".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_AWS_S3_SECRET_KEY.to_string(),
+ "wrong-secret-key".to_string(),
+ );
+ envs
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/delta/mod.rs b/core/integration/tests/connectors/fixtures/delta/mod.rs
index e413f3427..6968f7c1a 100644
--- a/core/integration/tests/connectors/fixtures/delta/mod.rs
+++ b/core/integration/tests/connectors/fixtures/delta/mod.rs
@@ -17,4 +17,7 @@
mod fixture;
-pub use fixture::{DeltaFixture, DeltaS3Fixture};
+pub use fixture::{
+ DeltaCorruptedLogFixture, DeltaFixture, DeltaS3Fixture, DeltaS3NoBucketFixture,
+ DeltaS3NoTableFixture, DeltaS3WrongCredentialsFixture,
+};
diff --git a/core/integration/tests/connectors/fixtures/floci.rs b/core/integration/tests/connectors/fixtures/floci.rs
index 95baf1334..8bf4d04c1 100644
--- a/core/integration/tests/connectors/fixtures/floci.rs
+++ b/core/integration/tests/connectors/fixtures/floci.rs
@@ -59,6 +59,7 @@ impl FlociContainer {
.with_port(PORT.tcp())
.with_expected_status_code(200u16),
))
+ .with_env_var("FLOCI_SERVICES_S3_ENFORCE_AUTH", "true")
.with_container_name(container_name)
.with_mapped_port(0, PORT.tcp());
let request = if let Some(network) = network {
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index cddee53ab..7e5cedf5c 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -54,7 +54,10 @@ pub(crate) fn unique_container_name(service: &str) -> String {
pub use clickhouse::{
ClickHouseSinkFixture, ClickHouseSinkRowBinaryFixture, ClickHouseSinkStringFixture,
};
-pub use delta::{DeltaFixture, DeltaS3Fixture};
+pub use delta::{
+ DeltaCorruptedLogFixture, DeltaFixture, DeltaS3Fixture, DeltaS3NoBucketFixture,
+ DeltaS3NoTableFixture, DeltaS3WrongCredentialsFixture,
+};
pub use doris::{
DorisOps, DorisSinkColumnsMappingFixture, DorisSinkCsvFixture, DorisSinkFixture,
DorisSinkMaxFilterRatioFixture, DorisSinkPreCreatedFixture,