This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch iceberg_report_followups in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c234ec5b5a5c9ca7a151ddefc998b510b9a4e1c2 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Nov 3 15:44:59 2026 +0100 fix(connectors): address Iceberg sink report follow-ups --- Cargo.lock | 6 +- Cargo.toml | 3 +- README.md | 2 +- .../example_config/connectors/iceberg_sink.toml | 5 +- core/connectors/sdk/Cargo.toml | 3 +- core/connectors/sdk/src/log.rs | 27 +++++- core/connectors/sdk/src/sink.rs | 8 +- core/connectors/sdk/src/source.rs | 8 +- core/connectors/sdk/tests/plugin_logging.rs | 100 +++++++++++++++++++++ core/connectors/sinks/iceberg_sink/Cargo.toml | 3 +- core/connectors/sinks/iceberg_sink/README.md | 29 +++++- core/connectors/sinks/iceberg_sink/config.toml | 2 +- core/connectors/sinks/iceberg_sink/src/lib.rs | 9 +- core/connectors/sinks/iceberg_sink/src/props.rs | 31 +++++-- .../iceberg_sink/src/router/dynamic_router.rs | 22 +++-- .../sinks/iceberg_sink/src/router/mod.rs | 7 +- core/connectors/sinks/iceberg_sink/src/sink.rs | 7 +- core/server/config.toml | 1 + core/server/server.http | 17 ++-- 19 files changed, 234 insertions(+), 56 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 130f544a8..38cd9f10a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7590,7 +7590,7 @@ dependencies = [ [[package]] name = "iggy_connector_iceberg_sink" -version = "0.5.1-edge.1" +version = "0.5.1-edge.2" dependencies = [ "arrow-array 58.4.1", "arrow-json 58.4.1", @@ -7601,6 +7601,7 @@ dependencies = [ "iceberg-storage-opendal", "iggy_connector_sdk", "parquet 58.4.1", + "secrecy", "serde", "simd-json", "strum 0.28.0", @@ -7851,7 +7852,7 @@ dependencies = [ [[package]] name = "iggy_connector_sdk" -version = "0.5.1-edge.2" +version = "0.5.1-edge.3" dependencies = [ "anyhow", "apache-avro 0.22.0", @@ -7881,6 +7882,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "tracing", + "tracing-log", "tracing-subscriber", "uuid", "wiremock", diff --git a/Cargo.toml b/Cargo.toml index de21ccbe0..1ec1e85c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -220,7 +220,7 @@ iggy-gateway-kafka = { path = "gateways/kafka" } iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.1-edge.8" } iggy_common = { path = "core/common", version = "0.11.1-edge.10" } iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" } -iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.1-edge.2" } +iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.1-edge.3" } iggy_kafka_wire = { path = "core/kafka_wire" } indexmap = "2.14.2" integration = { path = "core/integration" } @@ -372,6 +372,7 @@ tonic-prost = "0.14.9" tower-http = { version = "0.7.1", features = ["add-extension", "cors", "trace"] } tracing = "0.1.44" tracing-appender = "0.2.5" +tracing-log = "0.2.0" tracing-opentelemetry = "0.34.0" tracing-subscriber = { version = "0.3.23", default-features = false, features = [ "fmt", diff --git a/README.md b/README.md index e21f035e7..ece44b5d2 100644 --- a/README.md +++ b/README.md @@ -289,7 +289,7 @@ What each policy survives in a cluster of `n` replicas. A partition gets the `pe - `persisted` keeps every acknowledged write through any pattern of crashes and restarts. It is needed to survive `q` or more replicas crashing at once, power loss of the whole cluster, or any crash in a one-replica cluster. - `replicated` keeps every acknowledged write as long as fewer than `q` replicas lose unflushed writes at the same time, where `q`, the replicas that acknowledge each write, is half the replicas rounded up but at most 3 (2 in a two-replica cluster). A replica that returns with that partition awaiting a state transfer counts as one of them: its log is not used, even if its WAL still holds ops. -- Graceful restarts flush the WAL first and lose nothing, one at a time or all together. A crash (SIGKILL, OOM kill) loses the WAL writes not yet handed to the OS, and power loss also loses what the OS had not written to disk. Under `replicated`, `q` or more such crashes at once can lose the most recently acknowledged writes. +- Graceful restarts flush the WAL first and lose nothing, one at a time or all together. After a crash (SIGKILL, OOM kill) or power loss, a replica recovers a `replicated` partition up to the last WAL frontier it published, at a checkpoint, a durable operation or a graceful shutdown, and discards later writes even if the OS held them. Under `replicated`, `q` or more such crashes at once can lose the most recently acknowledged writes. In a one-replica cluster, any crash can. - A restarted `replicated` replica accepts no writes and acknowledges no prepares until it joins a settled view: one a running primary sends it, or one that an election starts. Until then writes and polls sent to it, binary or HTTP, fail with `TransientNotAccepted` (HTTP 503), which clients retry. When no primary answers, the restarted replicas hold an election and mark their logs as possibly incomplete. A new view starts only from `n - q + 1` replicas whose logs are not marked (2 in a two-replica cluster, a majority up to six replicas), or once every replica has reported, and the partition stays unavailable until then. After a full-cluster restart that means every replica must be back. A replica whose partition awaits a state transfer still reports without a log while it is recovering; once it has joined a view it stays out of later elections until its transfer completes. A replica that never returns halts the partition indefinitely (no operator override yet). diff --git a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml index 0f1079614..6ca19aaee 100644 --- a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml +++ b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml @@ -27,7 +27,10 @@ verbose = false stream = "example_stream" topics = ["example_topic"] schema = "json" -batch_length = 100 +# Every batch is one Iceberg commit, so larger batches keep snapshots and commit times down. +batch_length = 1000 +min_count = 1000 +max_wait = "5s" consumer_group = "iceberg_sink_connector" # Local S3 example diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml index 0cef19204..328caa4fa 100644 --- a/core/connectors/sdk/Cargo.toml +++ b/core/connectors/sdk/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_connector_sdk" -version = "0.5.1-edge.2" +version = "0.5.1-edge.3" description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second." edition = "2024" license = "Apache-2.0" @@ -65,6 +65,7 @@ strum_macros = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } +tracing-log = { workspace = true } tracing-subscriber = { workspace = true } uuid = { workspace = true } diff --git a/core/connectors/sdk/src/log.rs b/core/connectors/sdk/src/log.rs index 6741b8cce..4390e5751 100644 --- a/core/connectors/sdk/src/log.rs +++ b/core/connectors/sdk/src/log.rs @@ -16,8 +16,12 @@ // under the License. use tracing::field::{Field, Visit}; +use tracing::level_filters::LevelFilter; use tracing::{Event, Level, Subscriber}; -use tracing_subscriber::layer::{Context, Layer}; +use tracing_log::{AsLog, LogTracer, NormalizeEvent}; +use tracing_subscriber::layer::{Context, Layer, SubscriberExt}; +use tracing_subscriber::util::SubscriberInitExt; +use tracing_subscriber::{EnvFilter, Registry}; /// Callback for plugins to send log records to the runtime. /// @@ -37,6 +41,20 @@ pub type LogCallback = extern "C" fn( /// A tracing Layer that forwards events to the runtime via callback. /// This allows plugin logs to be routed through the runtime's OTEL subscriber. +/// Installs the subscriber of a plugin, which hands every event to the runtime through +/// `callback`, and routes the `log` records of the plugin's dependencies into it. Only the first +/// call of a plugin library takes effect. +pub(crate) fn init_plugin_logging(callback: LogCallback) { + _ = Registry::default() + .with(CallbackLayer::new(callback)) + .with(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO"))) + .try_init(); + // Each plugin library has its own `log` logger, and records sent to an unset one vanish. + _ = LogTracer::builder() + .with_max_level(LevelFilter::current().as_log()) + .init(); +} + pub struct CallbackLayer { callback: LogCallback, } @@ -49,7 +67,10 @@ impl CallbackLayer { impl<S: Subscriber> Layer<S> for CallbackLayer { fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { - let level = match *event.metadata().level() { + // A record bridged from `log` keeps its own target and level here. + let normalized = event.normalized_metadata(); + let metadata = normalized.as_ref().unwrap_or_else(|| event.metadata()); + let level = match *metadata.level() { Level::TRACE => 0u8, Level::DEBUG => 1, Level::INFO => 2, @@ -57,7 +78,7 @@ impl<S: Subscriber> Layer<S> for CallbackLayer { Level::ERROR => 4, }; - let target = event.metadata().target(); + let target = metadata.target(); let mut visitor = MessageVisitor::default(); event.record(&mut visitor); diff --git a/core/connectors/sdk/src/sink.rs b/core/connectors/sdk/src/sink.rs index 9891d2243..8a4b63686 100644 --- a/core/connectors/sdk/src/sink.rs +++ b/core/connectors/sdk/src/sink.rs @@ -18,9 +18,8 @@ use serde::de::DeserializeOwned; use tokio::sync::watch; use tracing::{error, info}; -use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt}; -use crate::log::{CallbackLayer, LogCallback}; +use crate::log::{LogCallback, init_plugin_logging}; use crate::{ ConsumedMessage, MessagesMetadata, Payload, RawMessages, Sink, TopicMetadata, get_runtime, }; @@ -66,10 +65,7 @@ impl<T: Sink + std::fmt::Debug> SinkContainer<T> { C: DeserializeOwned, { unsafe { - _ = Registry::default() - .with(CallbackLayer::new(log_callback)) - .with(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO"))) - .try_init(); + init_plugin_logging(log_callback); let slice = std::slice::from_raw_parts(config_ptr, config_len); let Ok(config_str) = std::str::from_utf8(slice) else { error!("Failed to read configuration for sink connector with ID: {id}"); diff --git a/core/connectors/sdk/src/source.rs b/core/connectors/sdk/src/source.rs index 766f3f4fe..f5f32d481 100644 --- a/core/connectors/sdk/src/source.rs +++ b/core/connectors/sdk/src/source.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::log::{CallbackLayer, LogCallback}; +use crate::log::{LogCallback, init_plugin_logging}; use crate::retry::exponential_backoff; use crate::{ConnectorState, ProducedMessages, Source, get_runtime}; use serde::de::DeserializeOwned; @@ -30,7 +30,6 @@ use tokio::{ task::JoinHandle, }; use tracing::{error, info, warn}; -use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt}; #[repr(C)] pub struct RawMessage { @@ -256,10 +255,7 @@ impl<T: Source + std::fmt::Debug + 'static> SourceContainer<T> { C: DeserializeOwned, { unsafe { - _ = Registry::default() - .with(CallbackLayer::new(log_callback)) - .with(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO"))) - .try_init(); + init_plugin_logging(log_callback); let slice = std::slice::from_raw_parts(config_ptr, config_len); let Ok(config_str) = std::str::from_utf8(slice) else { error!("Failed to read configuration for source connector with ID: {id}"); diff --git a/core/connectors/sdk/tests/plugin_logging.rs b/core/connectors/sdk/tests/plugin_logging.rs new file mode 100644 index 000000000..b8ea351db --- /dev/null +++ b/core/connectors/sdk/tests/plugin_logging.rs @@ -0,0 +1,100 @@ +// 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. + +//! The logging of a plugin is process-wide, so it is tested in a binary of its own. + +use async_trait::async_trait; +use iggy_connector_sdk::sink::SinkContainer; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata}; +use std::sync::Mutex; + +const WARN: u8 = 3; + +/// Level, target and message of every event the plugin handed to the runtime. +static FORWARDED: Mutex<Vec<(u8, String, String)>> = Mutex::new(Vec::new()); + +extern "C" fn record_event( + level: u8, + target_ptr: *const u8, + target_len: usize, + message_ptr: *const u8, + message_len: usize, +) { + let (target, message) = unsafe { + ( + std::slice::from_raw_parts(target_ptr, target_len), + std::slice::from_raw_parts(message_ptr, message_len), + ) + }; + FORWARDED.lock().expect("recorder was poisoned").push(( + level, + String::from_utf8_lossy(target).into_owned(), + String::from_utf8_lossy(message).into_owned(), + )); +} + +#[derive(Debug)] +struct QuietSink; + +#[async_trait] +impl Sink for QuietSink { + async fn open(&mut self) -> Result<(), Error> { + Ok(()) + } + + async fn consume( + &self, + _topic_metadata: &TopicMetadata, + _messages_metadata: MessagesMetadata, + _messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + Ok(()) + } +} + +#[test] +fn given_an_open_plugin_when_a_dependency_logs_through_the_log_crate_should_forward_the_record() { + let mut container = SinkContainer::<QuietSink>::new(1); + let config = b"{}"; + let status = unsafe { + container.open( + 1, + config.as_ptr(), + config.len(), + record_event, + |_, _: serde_json::Value| QuietSink, + ) + }; + assert_eq!(status, 0, "the sink should open"); + + // How OpenDAL's retry layer reports a retry. + tracing_log::log::warn!(target: "opendal::layers::retry", "will retry Write after 1s"); + + let forwarded = FORWARDED.lock().expect("recorder was poisoned"); + assert!( + forwarded + .iter() + .any(|(level, target, message)| *level == WARN + && target == "opendal::layers::retry" + && message.contains("will retry Write")), + "the record should reach the runtime with its own target and level: {forwarded:?}" + ); +} diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml index a8ecc7a54..ff649a1bc 100644 --- a/core/connectors/sinks/iceberg_sink/Cargo.toml +++ b/core/connectors/sinks/iceberg_sink/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_connector_iceberg_sink" -version = "0.5.1-edge.1" +version = "0.5.1-edge.2" edition = "2024" license = "Apache-2.0" keywords = ["iggy", "messaging", "streaming"] @@ -44,6 +44,7 @@ iceberg-catalog-rest = { workspace = true } iceberg-storage-opendal = { workspace = true } iggy_connector_sdk = { workspace = true } parquet = { workspace = true } +secrecy = { workspace = true } serde = { workspace = true } simd-json = { workspace = true } strum = { workspace = true } diff --git a/core/connectors/sinks/iceberg_sink/README.md b/core/connectors/sinks/iceberg_sink/README.md index 093cf18d5..6e8fff2a8 100644 --- a/core/connectors/sinks/iceberg_sink/README.md +++ b/core/connectors/sinks/iceberg_sink/README.md @@ -68,8 +68,8 @@ The Iggy Iceberg Connector will skip messages in the following cases: - The table name has no namespace or contains an empty name component. - The message does not contain the field specified in the `dynamic_route_field` configuration option, or is not a JSON object. -A lookup failure skips the affected message without failing the batch. Each batch loads its -destination tables again. The route field remains in the row, but the writer ignores it if the +A lookup failure skips the affected message without failing the batch, and every skipped +message is logged as a warning with its offset. Each batch loads its destination tables again. The route field remains in the row, but the writer ignores it if the target schema has no matching column. ### Dynamic routing configuration example @@ -117,6 +117,31 @@ according to the table's retry properties. The plugin does not replay failed bat runtime records a plugin error and continues polling with consumer auto-commit, so messages can be lost after failed or skipped writes. Replaying committed rows can create duplicates. +## Commit Rate and Table Maintenance + +Each batch that the runtime hands to the sink becomes one append commit per destination table, +which adds a snapshot and a manifest to the table. Commits slow down as the table metadata grows, +so fewer, larger batches keep both the commit latency and the metadata small. Set the batch size on +the stream: + +```toml +[[streams]] +stream = "events" +topics = ["users"] +schema = "json" +batch_length = 1000 # most messages in one batch +min_count = 1000 # wait until this many messages are ready, +max_wait = "5s" # or this long, whichever comes first +``` + +At 2,000 messages per second, `batch_length = 100` creates about 20 snapshots per second, and the +values above about 2. A poll also returns once its reply reaches 1 MiB, and `max_wait` bounds how +long a message waits for its batch. + +The sink does not expire snapshots or rewrite manifests. Run that maintenance from an engine that +supports it, for example Spark's `expire_snapshots` and `rewrite_manifests` procedures, on a +schedule that keeps the time travel you need. Some managed catalogs run it automatically. + ## Source Compatibility Use `schema = "json"` on the stream. Each JSON object represents a table row; nested objects diff --git a/core/connectors/sinks/iceberg_sink/config.toml b/core/connectors/sinks/iceberg_sink/config.toml index a4fa55c21..4125c27b4 100644 --- a/core/connectors/sinks/iceberg_sink/config.toml +++ b/core/connectors/sinks/iceberg_sink/config.toml @@ -27,7 +27,7 @@ verbose = true stream = "test_stream" topics = ["test_topic"] schema = "json" -batch_length = 100 +batch_length = 1000 consumer_group = "iceberg_sink_connector" [plugin_config] diff --git a/core/connectors/sinks/iceberg_sink/src/lib.rs b/core/connectors/sinks/iceberg_sink/src/lib.rs index f61ba1cd8..9786435d8 100644 --- a/core/connectors/sinks/iceberg_sink/src/lib.rs +++ b/core/connectors/sinks/iceberg_sink/src/lib.rs @@ -17,6 +17,7 @@ use crate::router::Router; use iggy_connector_sdk::{Error, sink_connector}; +use secrecy::SecretString; use serde::{Deserialize, Serialize}; use strum::Display as StrumDisplay; @@ -50,7 +51,9 @@ pub struct IcebergSink { router: Option<Box<dyn Router>>, } -#[derive(Debug, Serialize, Deserialize)] +// `Deserialize` only: nothing re-serializes the config, and leaving `Serialize` off keeps the +// store credentials unserializable. +#[derive(Debug, Deserialize)] pub struct IcebergSinkConfig { #[serde(default)] pub tables: Vec<String>, @@ -61,8 +64,8 @@ pub struct IcebergSinkConfig { #[serde(default)] pub dynamic_route_field: String, pub store_url: String, - pub store_access_key_id: Option<String>, - pub store_secret_access_key: Option<String>, + pub store_access_key_id: Option<SecretString>, + pub store_secret_access_key: Option<SecretString>, pub store_region: String, pub store_class: IcebergSinkStoreClass, /// Path-style S3 addressing (`http://host/bucket/key`). MinIO-style endpoints diff --git a/core/connectors/sinks/iceberg_sink/src/props.rs b/core/connectors/sinks/iceberg_sink/src/props.rs index 4c1635aa4..029c822eb 100644 --- a/core/connectors/sinks/iceberg_sink/src/props.rs +++ b/core/connectors/sinks/iceberg_sink/src/props.rs @@ -16,6 +16,7 @@ // under the License. use super::{Error, IcebergSinkConfig, IcebergSinkStoreClass}; +use secrecy::ExposeSecret; use std::collections::HashMap; pub fn init_props(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, Error> { @@ -36,10 +37,13 @@ fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, E ); match (&config.store_access_key_id, &config.store_secret_access_key) { (Some(access_key_id), Some(secret_access_key)) => { - props.insert("s3.access-key-id".to_string(), access_key_id.clone()); + props.insert( + "s3.access-key-id".to_string(), + access_key_id.expose_secret().to_owned(), + ); props.insert( "s3.secret-access-key".to_string(), - secret_access_key.clone(), + secret_access_key.expose_secret().to_owned(), ); } (None, None) => {} @@ -54,6 +58,7 @@ fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, E mod tests { use super::*; use crate::{IcebergSinkConfig, IcebergSinkStoreClass, IcebergSinkTypes}; + use secrecy::SecretString; fn base_config() -> IcebergSinkConfig { IcebergSinkConfig { @@ -96,8 +101,8 @@ mod tests { #[test] fn test_get_props_s3_full_credentials() { let config = IcebergSinkConfig { - store_access_key_id: Some("admin".to_string()), - store_secret_access_key: Some("password".to_string()), + store_access_key_id: Some(SecretString::from("admin")), + store_secret_access_key: Some(SecretString::from("password")), ..base_config() }; let props = get_props_s3(&config).expect("Should succeed with full credentials"); @@ -107,10 +112,24 @@ mod tests { assert_eq!(props.get("s3.secret-access-key").unwrap(), "password"); } + #[test] + fn given_store_credentials_when_the_config_is_debug_printed_should_redact_them() { + let config = IcebergSinkConfig { + store_access_key_id: Some(SecretString::from("AKIAEXAMPLEKEY")), + store_secret_access_key: Some(SecretString::from("example-secret-value")), + ..base_config() + }; + + let printed = format!("{config:?}"); + + assert!(!printed.contains("AKIAEXAMPLEKEY"), "{printed}"); + assert!(!printed.contains("example-secret-value"), "{printed}"); + } + #[test] fn test_get_props_s3_partial_access_key() { let config = IcebergSinkConfig { - store_access_key_id: Some("admin".to_string()), + store_access_key_id: Some(SecretString::from("admin")), store_secret_access_key: None, ..base_config() }; @@ -124,7 +143,7 @@ mod tests { fn test_get_props_s3_partial_secret_key() { let config = IcebergSinkConfig { store_access_key_id: None, - store_secret_access_key: Some("password".to_string()), + store_secret_access_key: Some(SecretString::from("password")), ..base_config() }; assert!( diff --git a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs index 17a717bd9..6e1321971 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs @@ -92,10 +92,7 @@ impl DynamicRouter { /// holding JSON arrives here as `Payload::Json`; proto text that is not JSON /// holds no document and does not route, and neither does any other variant. fn route_value(payload: &Payload, route_field: &str) -> Option<String> { - let Some(document) = payload.json_document() else { - warn!("Unsupported format for iceberg connector"); - return None; - }; + let document = payload.json_document()?; document .as_object() .and_then(|obj| obj.get(route_field)) @@ -114,9 +111,12 @@ impl Router for DynamicRouter { // Normalised once here so the routing read and the later write both // work off the document, rather than parsing the same proto text twice. message.payload = message.payload.into_json_document(); - let route_field_val = match self.extract_route_field(&message) { - Some(val) => val, - None => continue, + let Some(route_field_val) = self.extract_route_field(&message) else { + warn!( + "Dropping the message at offset {}: it is not a JSON object with the route field: {}", + message.offset, self.route_field + ); + continue; }; let message = match writer.push_to_existing(&route_field_val, message) { @@ -126,7 +126,8 @@ impl Router for DynamicRouter { if !is_valid_namespaced_table(&route_field_val) { warn!( - "Found invalid route field name on message: {route_field_val}. Route fields should have at least 1 namespace separated by '.' character before the table" + "Dropping the message at offset {}: its route field value: {route_field_val} names no table. It needs at least one namespace before the table, separated by '.'", + message.offset ); continue; } @@ -146,6 +147,11 @@ impl Router for DynamicRouter { .table_to_message .insert(route_field_val_cloned, message_vec); } + } else { + warn!( + "Dropping the message at offset {}: table: {route_field_val} does not load from the catalog", + message.offset + ); } } diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index 43fecfbfd..b6125f1fc 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -87,7 +87,12 @@ async fn write_data( // committed snapshot would corrupt the table. let table_commit = Transaction::new(table); - let action = table_commit.fast_append().add_data_files(data_files); + // The check reads every manifest of the table at once on each commit, and the data files + // carry fresh UUID names it could never match. + let action = table_commit + .fast_append() + .with_check_duplicate(false) + .add_data_files(data_files); let tx = action.apply(table_commit).map_err(|err| { let chain = format_error_chain(&err); diff --git a/core/connectors/sinks/iceberg_sink/src/sink.rs b/core/connectors/sinks/iceberg_sink/src/sink.rs index 1dd33cd39..a70d84fdb 100644 --- a/core/connectors/sinks/iceberg_sink/src/sink.rs +++ b/core/connectors/sinks/iceberg_sink/src/sink.rs @@ -42,12 +42,9 @@ impl Sink for IcebergSink { &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>(); + (Some(_), Some(_)) => { info!( - "Opened Iceberg sink connector with ID: {} for URL: {}, store access key ID: {redacted_store_key}*** store secret: {redacted_store_secret}***", + "Opened Iceberg sink connector with ID: {} for URL: {} with the configured store credentials", self.id, self.config.uri ); } diff --git a/core/server/config.toml b/core/server/config.toml index ac9dfb700..7d94aa7e0 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -1002,6 +1002,7 @@ change_log_activity_capacity = 16384 # Persistence barriers can also persist co-batched messages. # Persisted prepares are limited to 64 MiB including the consensus header. Oversized operations are rejected permanently. # Both policies write data to storage. "replicated" waits for VSR commit and application without an additional stable-storage barrier. +# After a crash, a "replicated" partition recovers up to the last WAL frontier it published: at checkpoints, durable operations and graceful shutdown. # "persisted" also requires recoverable stable-storage copies at the replication quorum before explicit success. # A singleton quorum is one replica. Its persisted completion follows recoverable local persistence. # Poll auto-commit remains asynchronous. A successful poll does not confirm that its submitted offset operation committed or became durable. diff --git a/core/server/server.http b/core/server/server.http index cbf4ea071..12ad7ff0f 100644 --- a/core/server/server.http +++ b/core/server/server.http @@ -29,8 +29,8 @@ @message_2_payload_base64 = d29ybGQ= @header_key_1_base64 = a2V5XzE= @header_value_1_base64 = dmFsdWUgMQ== -@header_key_2_base64 = Kg== -@header_value_2_base64 = AAAA +@header_key_2_base64 = KgAAAA== +@header_value_2_base64 = AAAAAA== @root_username = iggy @root_password = iggy @user1_username = user1 @@ -261,9 +261,11 @@ Content-Type: application/json { "name": "topic1", "partitions_count": 1, - "compression_algorithm": "none", - "max_topic_size": 0, - "message_expiry": 0 + "options": { + "compression_algorithm": "none", + "message_expiry": "7 days", + "max_topic_size": "1 GiB" + } } ### @@ -273,9 +275,8 @@ Content-Type: application/json { "name": "topic1", - "compression_algorithm": "none", - "max_topic_size": 0, - "message_expiry": 0 + "message_expiry": "1 day", + "max_topic_size": "2 GiB" } ###
