hubcio commented on code in PR #4759: URL: https://github.com/apache/iggy/pull/4759#discussion_r4482755439
########## core/connectors/sinks/iceberg_sink/src/router/redelivery.rs: ########## @@ -0,0 +1,385 @@ +// 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. + +//! Which messages of a partition a table already holds. At least once, the runtime hands a +//! message over again when it cannot prove it delivered, for example after a crash between the +//! commit and the offset store, or after a batch committed to one table and failed on another. +//! Every append records the last message it holds in its snapshot summary, and a delivery of +//! the same messages again appends only the ones after the newest append. + +use crate::router::BatchOrigin; +use crate::router::commit::table_load_error; +use dashmap::DashMap; +use iceberg::spec::Snapshot; +use iceberg::table::Table; +use iceberg::util::snapshot::ancestors_of; +use iceberg::{Catalog, TableIdent}; +use iggy_connector_sdk::{ConsumedMessage, Error}; +use std::collections::{BTreeMap, HashMap}; +use tracing::{info, warn}; +use uuid::Uuid; + +const STREAM_PROPERTY: &str = "iggy.stream"; +const TOPIC_PROPERTY: &str = "iggy.topic"; +const PARTITION_ID_PROPERTY: &str = "iggy.partition-id"; +const LAST_OFFSET_PROPERTY: &str = "iggy.last-offset"; +const LAST_MESSAGE_ID_PROPERTY: &str = "iggy.last-message-id"; +const LAST_TIMESTAMP_PROPERTY: &str = "iggy.last-timestamp"; + +/// The last message of a partition that an append holds. The message ID and the time the +/// server stored the message tell it apart from another message at the same offset in a later +/// history of the partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LastMessage { + offset: u64, + id: u128, + timestamp: u64, +} + +impl From<&ConsumedMessage> for LastMessage { + fn from(message: &ConsumedMessage) -> Self { + Self { + offset: message.offset, + id: message.id, + timestamp: message.timestamp, + } + } +} + +impl LastMessage { + fn from_properties(properties: &HashMap<String, String>) -> Option<Self> { + Some(Self { + offset: properties.get(LAST_OFFSET_PROPERTY)?.parse().ok()?, + id: properties.get(LAST_MESSAGE_ID_PROPERTY)?.parse().ok()?, + timestamp: properties.get(LAST_TIMESTAMP_PROPERTY)?.parse().ok()?, + }) + } +} + +/// The snapshot summary properties that record `last` as the last message of `origin` the +/// append holds. +pub(crate) fn snapshot_properties( + origin: &BatchOrigin<'_>, + last: LastMessage, +) -> HashMap<String, String> { + HashMap::from([ + (STREAM_PROPERTY.to_owned(), origin.stream.to_owned()), + (TOPIC_PROPERTY.to_owned(), origin.topic.to_owned()), + ( + PARTITION_ID_PROPERTY.to_owned(), + origin.partition_id.to_string(), + ), + (LAST_OFFSET_PROPERTY.to_owned(), last.offset.to_string()), + (LAST_MESSAGE_ID_PROPERTY.to_owned(), last.id.to_string()), + ( + LAST_TIMESTAMP_PROPERTY.to_owned(), + last.timestamp.to_string(), + ), + ]) +} + +/// A partition delivered to a table. The table UUID tells a table recreated under the same +/// name apart. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TablePartition { + table: Uuid, + stream: String, + topic: String, + partition_id: u32, +} + +impl TablePartition { + fn new(origin: &BatchOrigin<'_>, table: &Table) -> Self { + Self { + table: table.metadata().uuid(), + stream: origin.stream.to_owned(), + topic: origin.topic.to_owned(), + partition_id: origin.partition_id, + } + } +} + +/// The appends of one partition that a table holds. +#[derive(Debug, Default)] +struct Appends { + /// The last message of each append, by offset. Where two histories of the partition + /// appended at the same offset, the newer one is kept. + last_messages: BTreeMap<u64, LastMessage>, + newest: Option<LastMessage>, +} + +impl Appends { + fn record(&mut self, last: LastMessage) { Review Comment: warning: at least once, `record` adds a `last_messages` entry per commit and nothing trims it in steady state, so memory grows for the life of the sink. cap it, e.g. drop entries below the offset `Sink::flush` already confirmed. ########## core/connectors/sinks/iceberg_sink/src/router/static_router.rs: ########## @@ -39,44 +68,80 @@ impl StaticRouter { invalid_rows: InvalidRows, commit_retry: RetryPolicy, ) -> Result<Self, Error> { - let mut tables: Vec<Table> = Vec::with_capacity(declared_tables.len()); + let mut tables = Vec::with_capacity(declared_tables.len()); + let mut invalid_names = Vec::new(); for declared_table in declared_tables { if !is_valid_namespaced_table(declared_table) { error!( "Declared table {} is not valid. It has to include at least one namespace before the table name separated by '.' character", declared_table ); + invalid_names.push(declared_table.clone()); continue; } - let table = match table_exists(declared_table, catalog.as_ref()).await { - Some(table) => table, - None => { + let table = match load_table(declared_table, catalog.as_ref()).await { + Ok(table) => Some(Arc::new(table)), + Err(err) => { warn!( - "Declared table {} doesn't exist in the configured catalog. Skipping...", - declared_table + "Declared table {declared_table} does not load from the configured catalog, so each batch tries again. {}", + format_error_chain(&err) ); - continue; + None } }; - tables.push(table); + tables.push(DeclaredTable { + name: declared_table.clone(), + table: Mutex::new(table), + }); } + let loaded = tables + .iter() + .filter(|declared| declared.loaded().is_some()) + .count(); info!( - "Static router found {} tables on iceberg catalog from {} tables declared", - tables.len(), + "Static router found {loaded} tables on iceberg catalog from {} tables declared", declared_tables.len() ); - if tables.is_empty() { + if loaded == 0 { error!("No valid tables found. Can't initiate Iceberg connector"); return Err(Error::InvalidConfig); } Ok(StaticRouter { tables, + invalid_names, catalog, invalid_rows, commit_retry, + held: None, }) } + + /// The table of `declared`, loaded from the catalog when no earlier batch could. At most + /// once, a table that does not load is skipped for this batch. At least once, it fails the + /// batch, so that its messages are delivered again instead of lost. + async fn table(&self, declared: &DeclaredTable) -> Result<Option<Arc<Table>>, Error> { + if let Some(table) = declared.loaded() { + return Ok(Some(table)); + } + match load_table(&declared.name, self.catalog.as_ref()).await { + Ok(table) => { + let table = Arc::new(table); + info!("Declared table {} loaded from the catalog", declared.name); + *declared.lock() = Some(Arc::clone(&table)); + Ok(Some(table)) + } + Err(err) if self.held.is_some() => Err(table_load_error(&declared.name, &err)), + Err(err) => { Review Comment: warning: at most once, a declared table that fails to load now costs a catalog call and a warning on every batch, where before it was dropped once at startup. retry it on a backoff, or drop it like before. ########## core/connectors/sinks/iceberg_sink/src/router/mod.rs: ########## @@ -161,28 +237,35 @@ async fn write_data_files( let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); // Proto text holding JSON is the descriptor-less `proto_convert` fallback - // and is written as the document it holds. Proto text that is not JSON is - // skipped like any other payload without a document. - let documents: Vec<(u64, Cow<'_, simd_json::OwnedValue>)> = messages - .iter() - .filter_map(|message| { - let document = message.payload.json_document(); - if document.is_none() { - warn!( - "Unsupported type of payload, expected JSON, got {}", - messages_schema.to_string() - ); - } - document.map(|document| (message.offset, document)) - }) - .collect(); + // and is written as the document it holds. Proto text that is not JSON holds + // no document, so it is an invalid row like any other payload without one. + let mut rejection = None; + let mut documents: Vec<(u64, Cow<'_, simd_json::OwnedValue>)> = + Vec::with_capacity(messages.len()); + for message in messages { + if let Some(document) = message.payload.json_document() { + documents.push((message.offset, document)); + continue; + } + let reason = format!( + "The message at offset: {} ({origin}) holds no JSON document for table: {}, and the Iceberg sink requires schema = json, got {messages_schema}", + message.offset, + table.identifier() + ); + if invalid_rows == InvalidRows::Skip { + warn!("Skipping a row. {reason}"); + continue; + } + error!("{reason}"); Review Comment: warning: with the default `invalid_rows = "fail"`, at most once now stops at the first proto text without JSON and loses the rest of the run - before, only that message was skipped. skip it like before, or document the loss. ########## core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs: ########## @@ -76,17 +107,31 @@ impl Router for DynamicRouter { messages: Vec<ConsumedMessage>, ) -> Result<(), crate::Error> { let mut tables: HashMap<String, (Table, Vec<ConsumedMessage>)> = HashMap::new(); + // Route values that name no table, with the reason, so that a batch loads each once. + let mut unroutable: HashMap<String, String> = HashMap::new(); + let mut rejection = None; for mut message in messages { // 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 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 + let reason = format!( + "it is not a JSON object with the route field: {}", + self.route_field ); + rejection = self.unroutable(origin, &message, &reason); + if rejection.is_some() { + break; + } continue; }; + if let Some(reason) = unroutable.get(&route_field_val) { + rejection = self.unroutable(origin, &message, reason); + if rejection.is_some() { + break; + } + continue; + } let entry = match tables.entry(route_field_val) { Review Comment: warning: writes are per route value but held markers are per table UUID, so two route values that resolve to one table (e.g. a case-insensitive catalog) can make a redelivery skip a message. group messages by loaded table UUID instead. ########## core/connectors/sinks/iceberg_sink/README.md: ########## @@ -50,7 +51,7 @@ max_retry_delay = "30s" - **store_region**: The region of the object storage. Required, including for S3-compatible stores that ignore it. - **store_class**: The storage class to use. **Currently, only S3-compatible storage is supported.** - **store_path_style_access**: Use path-style S3 URLs (`http://host/bucket/key`). Defaults to `true`, which MinIO-style endpoints require; set to `false` for stores that only accept virtual-hosted-style URLs. -- **invalid_rows**: What to do with a row that does not match the table schema. `fail`, the default, fails the whole batch. `skip` writes the rows that match and logs each skipped row as a warning with its offset. +- **invalid_rows**: What to do with a row that does not match the table schema. `fail`, the default, fails the whole batch. `skip` writes the rows that match and logs each skipped row as a warning with its offset. At least once, it also decides what happens to a message that dynamic routing cannot route. Review Comment: warning: this says `fail` fails the whole batch, but the sink now commits the rows before the first rejected message, then fails the call. also at lines 233-234 and on `InvalidRows::Fail` in `src/lib.rs:43`. ########## core/connectors/sinks/iceberg_sink/src/router/redelivery.rs: ########## @@ -0,0 +1,385 @@ +// 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. + +//! Which messages of a partition a table already holds. At least once, the runtime hands a +//! message over again when it cannot prove it delivered, for example after a crash between the +//! commit and the offset store, or after a batch committed to one table and failed on another. +//! Every append records the last message it holds in its snapshot summary, and a delivery of +//! the same messages again appends only the ones after the newest append. + +use crate::router::BatchOrigin; +use crate::router::commit::table_load_error; +use dashmap::DashMap; +use iceberg::spec::Snapshot; +use iceberg::table::Table; +use iceberg::util::snapshot::ancestors_of; +use iceberg::{Catalog, TableIdent}; +use iggy_connector_sdk::{ConsumedMessage, Error}; +use std::collections::{BTreeMap, HashMap}; +use tracing::{info, warn}; +use uuid::Uuid; + +const STREAM_PROPERTY: &str = "iggy.stream"; +const TOPIC_PROPERTY: &str = "iggy.topic"; +const PARTITION_ID_PROPERTY: &str = "iggy.partition-id"; +const LAST_OFFSET_PROPERTY: &str = "iggy.last-offset"; +const LAST_MESSAGE_ID_PROPERTY: &str = "iggy.last-message-id"; +const LAST_TIMESTAMP_PROPERTY: &str = "iggy.last-timestamp"; + +/// The last message of a partition that an append holds. The message ID and the time the +/// server stored the message tell it apart from another message at the same offset in a later +/// history of the partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LastMessage { + offset: u64, + id: u128, + timestamp: u64, +} + +impl From<&ConsumedMessage> for LastMessage { + fn from(message: &ConsumedMessage) -> Self { + Self { + offset: message.offset, + id: message.id, + timestamp: message.timestamp, + } + } +} + +impl LastMessage { + fn from_properties(properties: &HashMap<String, String>) -> Option<Self> { + Some(Self { + offset: properties.get(LAST_OFFSET_PROPERTY)?.parse().ok()?, + id: properties.get(LAST_MESSAGE_ID_PROPERTY)?.parse().ok()?, + timestamp: properties.get(LAST_TIMESTAMP_PROPERTY)?.parse().ok()?, + }) + } +} + +/// The snapshot summary properties that record `last` as the last message of `origin` the +/// append holds. +pub(crate) fn snapshot_properties( + origin: &BatchOrigin<'_>, + last: LastMessage, +) -> HashMap<String, String> { + HashMap::from([ + (STREAM_PROPERTY.to_owned(), origin.stream.to_owned()), + (TOPIC_PROPERTY.to_owned(), origin.topic.to_owned()), + ( + PARTITION_ID_PROPERTY.to_owned(), + origin.partition_id.to_string(), + ), + (LAST_OFFSET_PROPERTY.to_owned(), last.offset.to_string()), + (LAST_MESSAGE_ID_PROPERTY.to_owned(), last.id.to_string()), + ( + LAST_TIMESTAMP_PROPERTY.to_owned(), + last.timestamp.to_string(), + ), + ]) +} + +/// A partition delivered to a table. The table UUID tells a table recreated under the same +/// name apart. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TablePartition { + table: Uuid, + stream: String, + topic: String, + partition_id: u32, +} + +impl TablePartition { + fn new(origin: &BatchOrigin<'_>, table: &Table) -> Self { + Self { + table: table.metadata().uuid(), + stream: origin.stream.to_owned(), + topic: origin.topic.to_owned(), + partition_id: origin.partition_id, + } + } +} + +/// The appends of one partition that a table holds. +#[derive(Debug, Default)] +struct Appends { + /// The last message of each append, by offset. Where two histories of the partition + /// appended at the same offset, the newer one is kept. + last_messages: BTreeMap<u64, LastMessage>, + newest: Option<LastMessage>, +} + +impl Appends { + fn record(&mut self, last: LastMessage) { + self.last_messages.insert(last.offset, last); + self.newest = Some(last); + } +} + +/// What the sink knows each table holds of each partition it delivers. A partition is unknown +/// until the sink reads the table's snapshot history, and again after a commit that failed, +/// since the catalog may have applied it. +#[derive(Debug, Default)] +pub(crate) struct HeldMessages { + known: DashMap<TablePartition, Appends>, +} + +impl HeldMessages { + /// The tail of `messages` that `table` does not hold yet. `current` is a copy of `table` + /// just loaded from the catalog, which saves loading it again. + pub(crate) async fn unheld<'m>( + &self, + origin: &BatchOrigin<'_>, + table: &Table, + current: Option<&Table>, + catalog: &dyn Catalog, + messages: &'m [ConsumedMessage], + ) -> Result<&'m [ConsumedMessage], Error> { + let key = TablePartition::new(origin, table); + if !self.known.contains_key(&key) { + let appends = match current { + Some(current) => appends_in(current, origin), + None => { + let identifier = table.identifier(); + let current = catalog + .load_table(identifier) + .await + .map_err(|error| table_load_error(&identifier.to_string(), &error))?; + appends_in(¤t, origin) + } + }; + self.known.entry(key.clone()).or_insert(appends); + } + let held = match self.known.get_mut(&key) { + Some(mut appends) => { + match held_prefix(origin, table.identifier(), messages, &appends) { + Prefix::Held(held) => held, + Prefix::NewHistory => { + *appends = Appends::default(); + 0 + } + Prefix::Unproven => 0, + } + } + None => 0, + }; + Ok(&messages[held..]) + } + + /// The last message of a commit to record in its snapshot. A batch that the table could not + /// prove it held, but that ends below the newest append, records that append again, so that + /// the record never moves back within one history of the partition. + pub(crate) fn last_to_record( + &self, + origin: &BatchOrigin<'_>, + table: &Table, + last: LastMessage, + ) -> LastMessage { + self.known + .get(&TablePartition::new(origin, table)) + .and_then(|appends| appends.newest) + .filter(|newest| newest.offset > last.offset) + .unwrap_or(last) + } + + pub(crate) fn committed(&self, origin: &BatchOrigin<'_>, table: &Table, last: LastMessage) { + self.known + .entry(TablePartition::new(origin, table)) + .or_default() + .record(last); + } + + pub(crate) fn forget(&self, origin: &BatchOrigin<'_>, table: &Table) { + self.known.remove(&TablePartition::new(origin, table)); + } +} + +/// The appends of `origin` on the current snapshot's ancestry. Snapshots that a rollback or a +/// replace left off that ancestry do not count, since the table no longer holds their rows. +fn appends_in(table: &Table, origin: &BatchOrigin<'_>) -> Appends { + let mut appends = Appends::default(); + let Some(current) = table.metadata().current_snapshot_id() else { + return appends; + }; + let partition_id = origin.partition_id.to_string(); + for snapshot in ancestors_of(&table.metadata_ref(), current) { + if !records(&snapshot, origin, &partition_id) { + continue; + } + let properties = &snapshot.summary().additional_properties; + let Some(last) = LastMessage::from_properties(properties) else { + warn!( + "Iceberg table {} has snapshot {} of {origin} without a readable last message. Skipping it", + table.identifier(), + snapshot.snapshot_id() + ); + continue; + }; + // The walk runs from the newest snapshot back, so the first one seen at an offset wins. + appends.last_messages.entry(last.offset).or_insert(last); + appends.newest.get_or_insert(last); + } + appends +} + +fn records(snapshot: &Snapshot, origin: &BatchOrigin<'_>, partition_id: &str) -> bool { + let properties = &snapshot.summary().additional_properties; + properties.get(STREAM_PROPERTY).map(String::as_str) == Some(origin.stream) + && properties.get(TOPIC_PROPERTY).map(String::as_str) == Some(origin.topic) + && properties.get(PARTITION_ID_PROPERTY).map(String::as_str) == Some(partition_id) +} + +/// What a table holds of the start of a batch. +#[derive(Debug, PartialEq, Eq)] +enum Prefix { + /// This many messages, proven to be the history the table holds. + Held(usize), + /// None: the table holds other messages at the batch's offsets. + NewHistory, + /// None that can be proven, although the batch reaches offsets the table holds. + Unproven, +} + +/// What the table holds of the start of `messages`. Messages count as held only when the batch +/// carries the last message of one of the table's appends, which proves that it is the history +/// the table holds. Then every message up to the newest append is held. +fn held_prefix( + origin: &BatchOrigin<'_>, + table: &TableIdent, + messages: &[ConsumedMessage], + appends: &Appends, +) -> Prefix { + let Some(newest) = appends.newest else { + return Prefix::Held(0); + }; + // A run of one partition carries ascending offsets. + let held = messages.partition_point(|message| message.offset <= newest.offset); + if held == 0 { + return Prefix::Held(0); + } + let mut other_history = false; + for message in &messages[..held] { + match appends.last_messages.get(&message.offset) { + Some(last) if *last == LastMessage::from(message) => { + info!( + "Skipping {held} messages from offset: {} to offset: {} ({origin}) that Iceberg table {table} already holds", + messages[0].offset, + messages[held - 1].offset, + ); + return Prefix::Held(held); Review Comment: warning: one matched record is treated as proof that every message below `newest` landed, so an at-least-once replay skips rows that an earlier at-most-once run lost. record each append's first offset and skip only inside a contiguous run of appends. ########## core/connectors/sinks/iceberg_sink/src/router/redelivery.rs: ########## @@ -0,0 +1,385 @@ +// 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. + +//! Which messages of a partition a table already holds. At least once, the runtime hands a +//! message over again when it cannot prove it delivered, for example after a crash between the +//! commit and the offset store, or after a batch committed to one table and failed on another. +//! Every append records the last message it holds in its snapshot summary, and a delivery of +//! the same messages again appends only the ones after the newest append. + +use crate::router::BatchOrigin; +use crate::router::commit::table_load_error; +use dashmap::DashMap; +use iceberg::spec::Snapshot; +use iceberg::table::Table; +use iceberg::util::snapshot::ancestors_of; +use iceberg::{Catalog, TableIdent}; +use iggy_connector_sdk::{ConsumedMessage, Error}; +use std::collections::{BTreeMap, HashMap}; +use tracing::{info, warn}; +use uuid::Uuid; + +const STREAM_PROPERTY: &str = "iggy.stream"; +const TOPIC_PROPERTY: &str = "iggy.topic"; +const PARTITION_ID_PROPERTY: &str = "iggy.partition-id"; +const LAST_OFFSET_PROPERTY: &str = "iggy.last-offset"; +const LAST_MESSAGE_ID_PROPERTY: &str = "iggy.last-message-id"; +const LAST_TIMESTAMP_PROPERTY: &str = "iggy.last-timestamp"; + +/// The last message of a partition that an append holds. The message ID and the time the +/// server stored the message tell it apart from another message at the same offset in a later +/// history of the partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LastMessage { + offset: u64, + id: u128, + timestamp: u64, +} + +impl From<&ConsumedMessage> for LastMessage { + fn from(message: &ConsumedMessage) -> Self { + Self { + offset: message.offset, + id: message.id, + timestamp: message.timestamp, + } + } +} + +impl LastMessage { + fn from_properties(properties: &HashMap<String, String>) -> Option<Self> { + Some(Self { + offset: properties.get(LAST_OFFSET_PROPERTY)?.parse().ok()?, + id: properties.get(LAST_MESSAGE_ID_PROPERTY)?.parse().ok()?, + timestamp: properties.get(LAST_TIMESTAMP_PROPERTY)?.parse().ok()?, + }) + } +} + +/// The snapshot summary properties that record `last` as the last message of `origin` the +/// append holds. +pub(crate) fn snapshot_properties( + origin: &BatchOrigin<'_>, + last: LastMessage, +) -> HashMap<String, String> { + HashMap::from([ + (STREAM_PROPERTY.to_owned(), origin.stream.to_owned()), + (TOPIC_PROPERTY.to_owned(), origin.topic.to_owned()), + ( + PARTITION_ID_PROPERTY.to_owned(), + origin.partition_id.to_string(), + ), + (LAST_OFFSET_PROPERTY.to_owned(), last.offset.to_string()), + (LAST_MESSAGE_ID_PROPERTY.to_owned(), last.id.to_string()), + ( + LAST_TIMESTAMP_PROPERTY.to_owned(), + last.timestamp.to_string(), + ), + ]) +} + +/// A partition delivered to a table. The table UUID tells a table recreated under the same Review Comment: warning: the static router keeps its startup `Table`, so a recreated table keeps the old UUID here and a redelivery can skip rows the new table never got. document that a recreated table needs a sink restart, like a rollback. ########## core/connectors/sinks/iceberg_sink/src/router/commit.rs: ########## @@ -491,17 +572,24 @@ mod tests { async fn given_an_unreachable_catalog_when_the_attempts_run_out_should_fail() { let (catalog, table) = FlakyCatalog::with_table(&[CommitFailure::Unreachable; 3]).await; - let error = commit_data_files(&table, &catalog, vec![data_file(&table)], FAST_RETRY) - .await - .expect_err("The commit should fail once the attempts run out"); + let error = commit_data_files( + &table, + &catalog, + vec![data_file(&table)], + HashMap::new(), + FAST_RETRY, + ) + .await + .expect_err("The commit should fail once the attempts run out"); assert!(matches!(error, Error::CatalogCommitError(_)), "{error}"); assert_eq!(catalog.update_calls(), FAST_RETRY.max_attempts); assert_eq!(catalog.snapshot_count(&table).await, 0); } /// A REST catalog whose server answers every request with `status` and no body. - async fn rest_catalog_answering(status: StatusCode) -> RestCatalog { + /// A REST catalog that answers every call with `status`. Review Comment: nit: this doc line repeats the one above it. ########## core/connectors/sinks/iceberg_sink/src/router/commit.rs: ########## @@ -82,17 +82,50 @@ pub(crate) async fn commit_data_files( failure.attempts, chain ); - Error::CatalogCommitError(chain) + // A conflict outlasted the table's own retries, so a delivery of the batch again rebuilds + // the transaction on fresh metadata. Any other refusal, such as a denied write, fails the + // same way every time. + if is_transient(&failure.error) || failure.error.kind() == ErrorKind::CatalogCommitConflicts + { + Error::CatalogCommitError(chain) + } else { + Error::PermanentHttpError(format!( + "The Iceberg catalog refused the commit to table {identifier}: {chain}" + )) + } }) } -/// A commit that could not reach the catalog, or that the catalog failed or throttled, can pass -/// on another attempt. A conflict is not retried here: `Transaction::commit` has already retried -/// it as far as the table's `commit.retry.*` properties allow. -fn is_transient(error: &iceberg::Error) -> bool { +/// A catalog call that could not reach the catalog, or that the catalog failed or throttled, can +/// pass on another attempt. A commit conflict is not retried here: `Transaction::commit` has +/// already retried it as far as the table's `commit.retry.*` properties allow. +pub(crate) fn is_transient(error: &iceberg::Error) -> bool { error.kind() == ErrorKind::Unexpected && response_status(error).is_none_or(is_transient_status) } +/// The error for a table that does not load from the catalog. An unreachable, failing or +/// throttling catalog is transient. Any other refusal, such as a missing table or denied access, +/// fails the same way until the configuration or the catalog changes. +pub(crate) fn table_load_error(table: &str, error: &iceberg::Error) -> Error { + let reason = format!( + "Failed to load Iceberg table {table} from the catalog: {}", + format_error_chain(error) + ); + if is_transient(error) { + Error::Connection(reason) + } else { + Error::InvalidConfigValue(reason) Review Comment: nit: a refused table load (e.g. 403) maps to `InvalidConfigValue` but a refused commit maps to `PermanentHttpError`. map the refused load to `PermanentHttpError` too and keep `InvalidConfigValue` for a missing declared table. ########## core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs: ########## @@ -47,12 +53,37 @@ impl DynamicRouter { route_field, invalid_rows, commit_retry, + held: None, } } fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String> { route_value(&message.payload, &self.route_field) } + + /// Drops a message that names no table it can be written to. At least once, it rejects the + /// message instead, unless `invalid_rows` skips rows, so that the call stops at it. + fn unroutable( + &self, + origin: &BatchOrigin<'_>, + message: &ConsumedMessage, + reason: &str, + ) -> Option<Rejection> { + if self.held.is_some() && self.invalid_rows == InvalidRows::Fail { + return Some(Rejection { + offset: message.offset, + error: Error::InvalidRecordValue(format!( + "The message at offset {} ({origin}) cannot be routed: {reason}", Review Comment: nit: this rejection says `offset 7` while `write_data_files` and `decode_rows` say `offset: 7`, so one log search misses the other. use `offset: {}` here and in the warning at line 82. ########## core/connectors/sinks/iceberg_sink/src/router/commit.rs: ########## @@ -528,4 +616,398 @@ mod tests { .await .expect("Failed to create the REST catalog") } + + const ORIGIN: BatchOrigin<'static> = BatchOrigin { + connector_id: 1, + stream: "stream", + topic: "topic", + partition_id: 2, + }; + + fn message(offset: u64) -> ConsumedMessage { + ConsumedMessage { + id: u128::from(offset) + 100, + offset, + checksum: 0, + timestamp: offset + 1_000, + origin_timestamp: 0, + headers: None, + payload: Payload::Raw(Vec::new()), + } + } + + fn messages(offsets: std::ops::RangeInclusive<u64>) -> Vec<ConsumedMessage> { + offsets.map(message).collect() + } + + fn json_messages(offsets: std::ops::RangeInclusive<u64>) -> Vec<ConsumedMessage> { + offsets + .map(|offset| ConsumedMessage { + payload: Payload::Json(simd_json::json!({ "id": offset })), + ..message(offset) + }) + .collect() + } + + fn offsets(messages: &[ConsumedMessage]) -> Vec<u64> { + messages.iter().map(|message| message.offset).collect() + } + + async fn commit_up_to( + catalog: &FlakyCatalog, + table: &Table, + origin: &BatchOrigin<'_>, + offset: u64, + ) { + let properties = snapshot_properties(origin, LastMessage::from(&message(offset))); + commit_data_files( + table, + catalog, + vec![data_file(table)], + properties, + FAST_RETRY, + ) + .await + .expect("The append should commit"); + } + + async fn current_table(catalog: &FlakyCatalog, table: &Table) -> Table { + catalog + .inner + .load_table(table.identifier()) + .await + .expect("Failed to load table") + } + + #[allow(clippy::too_many_arguments)] Review Comment: nit: `write` takes 6 arguments, under clippy's default limit of 7, so this `#[allow(clippy::too_many_arguments)]` does nothing. drop it. ########## core/connectors/sdk/src/lib.rs: ########## @@ -161,8 +161,10 @@ pub trait Sink: Send + Sync { /// Called once after [`Sink::open`] when the sink is configured for at-least-once delivery. /// /// Returning `Ok` promises that, from now on, a successful [`Sink::consume`] never drops a - /// message without failing the call, and that a successful [`Sink::flush`] proves that the - /// destination acknowledged every message of the covered partitions handed over so far. + /// message without failing the call, unless the sink's configuration explicitly tells it to Review Comment: nit: "such messages" is vague here - name them like the SDK README does: messages the sink cannot write. ########## core/connectors/sinks/iceberg_sink/src/router/mod.rs: ########## @@ -87,50 +88,125 @@ fn format_error_chain(err: &dyn std::error::Error) -> String { mod arrow_streamer; mod commit; pub mod dynamic_router; +mod redelivery; pub mod static_router; pub fn is_valid_namespaced_table(input: &str) -> bool { let parts: Vec<&str> = input.split('.').collect(); parts.len() >= 2 && parts.iter().all(|part| !part.is_empty()) } -async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Table> { - let sliced_table = slice_user_table(route_field_val); - let table_ident = TableIdent::from_strs(&sliced_table).ok()?; +async fn load_table(route_field_val: &str, catalog: &dyn Catalog) -> iceberg::Result<Table> { + let table_ident = TableIdent::from_strs(slice_user_table(route_field_val))?; + catalog.load_table(&table_ident).await +} + +/// A message that a table rejects with `invalid_rows = "fail"`. Every message of the call before +/// it is committed, so that storing its offset, as an operator does to move past it, gives up +/// this message alone. +#[derive(Debug)] +pub(crate) struct Rejection { + pub(crate) offset: u64, + pub(crate) error: Error, +} + +/// What a write of a call's messages to one table did. +#[derive(Debug, Default)] +pub(crate) struct Written { + /// The rows committed, which leaves out the rows the sink skipped. + pub(crate) committed_rows: u64, + pub(crate) rejection: Option<Rejection>, +} + +/// Rejections of several tables stop the call at the first rejected message. +fn first_rejection(rejection: Option<Rejection>, other: Option<Rejection>) -> Option<Rejection> { + match (rejection, other) { + (Some(rejection), Some(other)) if other.offset < rejection.offset => Some(other), + (rejection, other) => rejection.or(other), + } +} - catalog.load_table(&table_ident).await.ok() +/// The messages before `rejection`, which a call still has to deliver. +fn before<'m>( + messages: &'m [ConsumedMessage], + rejection: Option<&Rejection>, +) -> &'m [ConsumedMessage] { + match rejection { + Some(rejection) => { + &messages[..messages.partition_point(|message| message.offset < rejection.offset)] + } + None => messages, + } } -/// Returns the number of rows committed, which leaves out the rows the sink skipped. +/// Writes and commits `messages` to `table`. With `held`, the sink delivers at least once: the +/// messages that `table` already holds are not appended again, and the commit records the last +/// message it holds. `current` is a copy of `table` just loaded from the catalog. +#[allow(clippy::too_many_arguments)] async fn write_data( origin: &BatchOrigin<'_>, messages: &[ConsumedMessage], table: &Table, + current: Option<&Table>, Review Comment: simplification: every caller passes `current` as `None` or as `table` itself, so a `bool` that says `table` is fresh from the catalog is enough and cannot point at the wrong table. ########## core/connectors/sinks/iceberg_sink/src/router/static_router.rs: ########## @@ -16,20 +16,49 @@ // under the License. use crate::InvalidRows; -use crate::router::{BatchOrigin, Router, is_valid_namespaced_table, table_exists, write_data}; +use crate::router::commit::table_load_error; +use crate::router::redelivery::HeldMessages; +use crate::router::{ + BatchOrigin, Router, before, first_rejection, format_error_chain, is_valid_namespaced_table, + load_table, write_data, +}; use async_trait::async_trait; use iceberg::Catalog; use iceberg::table::Table; use iggy_connector_sdk::retry::RetryPolicy; use iggy_connector_sdk::{ConsumedMessage, Error, Schema}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use tracing::{error, info, warn}; +/// A table that the configuration declares. A table that does not load is loaded again by the +/// next batch, so that a catalog that was unreachable at startup does not drop it for good. +#[derive(Debug)] +struct DeclaredTable { Review Comment: simplification: the table is set once and never replaced, so `OnceLock<Table>` with `get`/`get_or_init` can replace `Mutex<Option<Arc<Table>>>` and the poison handling, and `table` can return `Option<&Table>`. ########## core/connectors/sinks/iceberg_sink/src/router/redelivery.rs: ########## @@ -0,0 +1,385 @@ +// 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. + +//! Which messages of a partition a table already holds. At least once, the runtime hands a +//! message over again when it cannot prove it delivered, for example after a crash between the +//! commit and the offset store, or after a batch committed to one table and failed on another. +//! Every append records the last message it holds in its snapshot summary, and a delivery of +//! the same messages again appends only the ones after the newest append. + +use crate::router::BatchOrigin; +use crate::router::commit::table_load_error; +use dashmap::DashMap; +use iceberg::spec::Snapshot; +use iceberg::table::Table; +use iceberg::util::snapshot::ancestors_of; +use iceberg::{Catalog, TableIdent}; +use iggy_connector_sdk::{ConsumedMessage, Error}; +use std::collections::{BTreeMap, HashMap}; +use tracing::{info, warn}; +use uuid::Uuid; + +const STREAM_PROPERTY: &str = "iggy.stream"; +const TOPIC_PROPERTY: &str = "iggy.topic"; +const PARTITION_ID_PROPERTY: &str = "iggy.partition-id"; +const LAST_OFFSET_PROPERTY: &str = "iggy.last-offset"; +const LAST_MESSAGE_ID_PROPERTY: &str = "iggy.last-message-id"; +const LAST_TIMESTAMP_PROPERTY: &str = "iggy.last-timestamp"; + +/// The last message of a partition that an append holds. The message ID and the time the +/// server stored the message tell it apart from another message at the same offset in a later +/// history of the partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LastMessage { + offset: u64, + id: u128, + timestamp: u64, +} + +impl From<&ConsumedMessage> for LastMessage { + fn from(message: &ConsumedMessage) -> Self { + Self { + offset: message.offset, + id: message.id, + timestamp: message.timestamp, + } + } +} + +impl LastMessage { + fn from_properties(properties: &HashMap<String, String>) -> Option<Self> { + Some(Self { + offset: properties.get(LAST_OFFSET_PROPERTY)?.parse().ok()?, + id: properties.get(LAST_MESSAGE_ID_PROPERTY)?.parse().ok()?, + timestamp: properties.get(LAST_TIMESTAMP_PROPERTY)?.parse().ok()?, + }) + } +} + +/// The snapshot summary properties that record `last` as the last message of `origin` the +/// append holds. +pub(crate) fn snapshot_properties( + origin: &BatchOrigin<'_>, + last: LastMessage, +) -> HashMap<String, String> { + HashMap::from([ + (STREAM_PROPERTY.to_owned(), origin.stream.to_owned()), + (TOPIC_PROPERTY.to_owned(), origin.topic.to_owned()), + ( + PARTITION_ID_PROPERTY.to_owned(), + origin.partition_id.to_string(), + ), + (LAST_OFFSET_PROPERTY.to_owned(), last.offset.to_string()), + (LAST_MESSAGE_ID_PROPERTY.to_owned(), last.id.to_string()), + ( + LAST_TIMESTAMP_PROPERTY.to_owned(), + last.timestamp.to_string(), + ), + ]) +} + +/// A partition delivered to a table. The table UUID tells a table recreated under the same +/// name apart. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TablePartition { + table: Uuid, + stream: String, + topic: String, + partition_id: u32, +} + +impl TablePartition { + fn new(origin: &BatchOrigin<'_>, table: &Table) -> Self { + Self { + table: table.metadata().uuid(), + stream: origin.stream.to_owned(), + topic: origin.topic.to_owned(), + partition_id: origin.partition_id, + } + } +} + +/// The appends of one partition that a table holds. +#[derive(Debug, Default)] +struct Appends { + /// The last message of each append, by offset. Where two histories of the partition + /// appended at the same offset, the newer one is kept. + last_messages: BTreeMap<u64, LastMessage>, + newest: Option<LastMessage>, +} + +impl Appends { + fn record(&mut self, last: LastMessage) { + self.last_messages.insert(last.offset, last); + self.newest = Some(last); + } +} + +/// What the sink knows each table holds of each partition it delivers. A partition is unknown +/// until the sink reads the table's snapshot history, and again after a commit that failed, +/// since the catalog may have applied it. +#[derive(Debug, Default)] +pub(crate) struct HeldMessages { + known: DashMap<TablePartition, Appends>, +} + +impl HeldMessages { + /// The tail of `messages` that `table` does not hold yet. `current` is a copy of `table` + /// just loaded from the catalog, which saves loading it again. + pub(crate) async fn unheld<'m>( + &self, + origin: &BatchOrigin<'_>, + table: &Table, + current: Option<&Table>, + catalog: &dyn Catalog, + messages: &'m [ConsumedMessage], + ) -> Result<&'m [ConsumedMessage], Error> { + let key = TablePartition::new(origin, table); + if !self.known.contains_key(&key) { + let appends = match current { + Some(current) => appends_in(current, origin), + None => { + let identifier = table.identifier(); + let current = catalog + .load_table(identifier) + .await + .map_err(|error| table_load_error(&identifier.to_string(), &error))?; + appends_in(¤t, origin) + } + }; + self.known.entry(key.clone()).or_insert(appends); Review Comment: simplification: `contains_key`, then `entry(key.clone()).or_insert`, then `get_mut` with a `None => 0` arm for a key that was just inserted. after the optional load, keep the guard from `entry(key).or_insert_with(..)` instead. ########## core/connectors/sinks/iceberg_sink/src/router/mod.rs: ########## @@ -87,50 +88,125 @@ fn format_error_chain(err: &dyn std::error::Error) -> String { mod arrow_streamer; mod commit; pub mod dynamic_router; +mod redelivery; pub mod static_router; pub fn is_valid_namespaced_table(input: &str) -> bool { let parts: Vec<&str> = input.split('.').collect(); parts.len() >= 2 && parts.iter().all(|part| !part.is_empty()) } -async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Table> { - let sliced_table = slice_user_table(route_field_val); - let table_ident = TableIdent::from_strs(&sliced_table).ok()?; +async fn load_table(route_field_val: &str, catalog: &dyn Catalog) -> iceberg::Result<Table> { + let table_ident = TableIdent::from_strs(slice_user_table(route_field_val))?; + catalog.load_table(&table_ident).await +} + +/// A message that a table rejects with `invalid_rows = "fail"`. Every message of the call before +/// it is committed, so that storing its offset, as an operator does to move past it, gives up +/// this message alone. +#[derive(Debug)] +pub(crate) struct Rejection { + pub(crate) offset: u64, + pub(crate) error: Error, +} + +/// What a write of a call's messages to one table did. +#[derive(Debug, Default)] +pub(crate) struct Written { + /// The rows committed, which leaves out the rows the sink skipped. + pub(crate) committed_rows: u64, + pub(crate) rejection: Option<Rejection>, +} + +/// Rejections of several tables stop the call at the first rejected message. +fn first_rejection(rejection: Option<Rejection>, other: Option<Rejection>) -> Option<Rejection> { + match (rejection, other) { + (Some(rejection), Some(other)) if other.offset < rejection.offset => Some(other), + (rejection, other) => rejection.or(other), + } +} - catalog.load_table(&table_ident).await.ok() +/// The messages before `rejection`, which a call still has to deliver. +fn before<'m>( + messages: &'m [ConsumedMessage], + rejection: Option<&Rejection>, +) -> &'m [ConsumedMessage] { + match rejection { + Some(rejection) => { + &messages[..messages.partition_point(|message| message.offset < rejection.offset)] + } + None => messages, + } } -/// Returns the number of rows committed, which leaves out the rows the sink skipped. +/// Writes and commits `messages` to `table`. With `held`, the sink delivers at least once: the +/// messages that `table` already holds are not appended again, and the commit records the last +/// message it holds. `current` is a copy of `table` just loaded from the catalog. +#[allow(clippy::too_many_arguments)] async fn write_data( origin: &BatchOrigin<'_>, messages: &[ConsumedMessage], table: &Table, + current: Option<&Table>, Review Comment: simplification: both routers carry the same `catalog`, `invalid_rows`, `commit_retry` and `held` fields - one shared struct passed here would also drop the `too_many_arguments` allow. ########## core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs: ########## @@ -76,17 +107,31 @@ impl Router for DynamicRouter { messages: Vec<ConsumedMessage>, ) -> Result<(), crate::Error> { let mut tables: HashMap<String, (Table, Vec<ConsumedMessage>)> = HashMap::new(); + // Route values that name no table, with the reason, so that a batch loads each once. + let mut unroutable: HashMap<String, String> = HashMap::new(); + let mut rejection = None; for mut message in messages { // 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 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 + let reason = format!( + "it is not a JSON object with the route field: {}", + self.route_field ); + rejection = self.unroutable(origin, &message, &reason); Review Comment: simplification: `rejection = self.unroutable(..)` plus a `break` repeats at three exits of this loop - compute the unroutable reason once per message and handle it in one place. ########## core/connectors/sinks/iceberg_sink/src/router/commit.rs: ########## @@ -528,4 +616,398 @@ mod tests { .await .expect("Failed to create the REST catalog") } + + const ORIGIN: BatchOrigin<'static> = BatchOrigin { + connector_id: 1, + stream: "stream", + topic: "topic", + partition_id: 2, + }; + + fn message(offset: u64) -> ConsumedMessage { Review Comment: simplification: `message`/`messages` here copy the `redelivery.rs` test helpers, and the router tests copy the `FlakyCatalog::with_table_properties` setup and the one-attempt `RetryPolicy`. define them once in `commit::tests`, which is already `pub(crate)`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
