numinnex commented on code in PR #4123: URL: https://github.com/apache/iggy/pull/4123#discussion_r4411491746
########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || operator.write(&path, buffer.clone()), Review Comment: critical: nothing bounds this write. reqwest is built without a timeout, no OpenDAL `TimeoutLayer` is added, and `retry_async` has no deadline, while `s3_sink` gets 60 s from rust-s3 (`bucket.rs:106`). A peer that accepts the PUT and never answers (keepalive only catches dead peers) keeps `consume` pending forever, status Running, no error. The runtime calls the plugin synchronously on a host worker and the pool has one worker per CPU, so once stalled topics reach the worker count the whole runtime stalls, and stop/SIGTERM wait on the SDK instance lock (`sdk/src/sink.rs:299-318`). Fix: per-attempt timeout scaled to body size, driven through `writer()` + `abort()`. It is only safe together with the byte cap and the atomic-dir fix below. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || operator.write(&path, buffer.clone()), + ) + .await + .map_err(|failure| { Review Comment: warning: once this returns `Err` the batch is gone. The runtime commits the offset at poll and only logs a failed `consume`, so after 3 attempts (about 3 s), or one permanent error (403, NoSuchBucket), the archive has a permanent gap. The runtime side predates this PR, but the README should say it plainly: at-most-once, gaps possible, no duplicates. Keep retries bounded: an unbounded loop would hold stop/SIGTERM for the whole outage. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, Review Comment: warning: `Operator::write` never calls `abort()` on failure, so with `atomic_write_dir` set every failed attempt leaves a full-batch temp file in `.iggy-opendal-tmp` that nothing sweeps. `is_temporary` also covers EXDEV and ENOSPC, so an atomic dir on another filesystem passes `open()`, and each batch then leaks 3 temp files before it is dropped. Fix: write through `writer()` and `abort()` on error, and check at open that the atomic dir sits on the same device as `root`. ########## core/connectors/sinks/opendal_sink/src/formatter.rs: ########## @@ -0,0 +1,339 @@ +// 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. + +use std::collections::BTreeMap; + +use base64::Engine; +use chrono::{DateTime, Utc}; +use iggy_common::{HeaderKey, HeaderKind, HeaderValue}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata}; +use serde::{Serialize, Serializer, ser::SerializeMap}; +use serde_json::value::RawValue; + +use crate::OutputFormat; + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Formatter { + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, +} + +impl Formatter { + pub(crate) const fn new( + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, + ) -> Self { + Self { + output_format, + include_metadata, + include_headers, + } + } + + pub(crate) const fn output_format(&self) -> OutputFormat { + self.output_format + } + + pub(crate) fn format_batch( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<Vec<u8>, Error> { + let mut output = Vec::with_capacity(self.batch_capacity(messages, topic_metadata)); + if self.output_format == OutputFormat::JsonArray { + output.push(b'['); + } + + for (index, message) in messages.iter().enumerate() { + if self.output_format == OutputFormat::JsonArray && index > 0 { + output.push(b','); + } + + match self.output_format { + OutputFormat::JsonLines | OutputFormat::JsonArray => self.write_json_message( + &mut output, + message, + topic_metadata, + messages_metadata, + )?, + OutputFormat::Raw => self.write_raw_payload(&mut output, message)?, + } + + if self.output_format == OutputFormat::JsonLines { + output.push(b'\n'); + } + } + + if self.output_format == OutputFormat::JsonArray { + output.push(b']'); + } + Ok(output) + } + + fn write_json_message( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<(), Error> { + let timestamp = self + .include_metadata + .then(|| timestamp_to_rfc3339(message.timestamp)); + let json_message = JsonMessage { + offset: self.include_metadata.then_some(message.offset), + timestamp: timestamp.as_deref(), + stream: self + .include_metadata + .then_some(topic_metadata.stream.as_str()), + topic: self + .include_metadata + .then_some(topic_metadata.topic.as_str()), + partition_id: self + .include_metadata + .then_some(messages_metadata.partition_id), + headers: if self.include_headers { + message.headers.as_ref().map(JsonHeaders) + } else { + None + }, + payload: JsonPayload(&message.payload), + }; + + serde_json::to_writer(output, &json_message).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize message at offset {}: {error}", + message.offset + )) + }) + } + + fn write_raw_payload( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + ) -> Result<(), Error> { + match &message.payload { + Payload::Json(value) => serde_json::to_writer(output, value).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize raw payload at offset {}: {error}", + message.offset + )) + }), + Payload::Raw(bytes) | Payload::FlatBuffer(bytes) | Payload::Avro(bytes) => { + output.extend_from_slice(bytes); + Ok(()) + } + Payload::Text(text) | Payload::Proto(text) => { + output.extend_from_slice(text.as_bytes()); + Ok(()) + } + } + } + + fn batch_capacity( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + ) -> usize { + let metadata_size = if self.include_metadata { + topic_metadata + .stream + .len() + .saturating_add(topic_metadata.topic.len()) + .saturating_add(128) + } else if self.output_format == OutputFormat::Raw { + 0 + } else { + 16 + }; + + messages.iter().fold(2, |capacity, message| { + capacity + .saturating_add(payload_size_hint(&message.payload)) + .saturating_add(metadata_size) + }) + } +} + +#[derive(Serialize)] +struct JsonMessage<'a> { + #[serde(skip_serializing_if = "Option::is_none")] + offset: Option<u64>, + #[serde(skip_serializing_if = "Option::is_none")] + timestamp: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + stream: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + topic: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + partition_id: Option<u32>, + #[serde(skip_serializing_if = "Option::is_none")] + headers: Option<JsonHeaders<'a>>, + payload: JsonPayload<'a>, +} + +struct JsonHeaders<'a>(&'a BTreeMap<HeaderKey, HeaderValue>); + +impl Serialize for JsonHeaders<'_> { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + let mut map = serializer.serialize_map(Some(self.0.len()))?; + for (key, value) in self.0 { + serialize_header(&mut map, key.as_str().unwrap_or(""), value)?; + } + map.end() + } +} + +struct JsonPayload<'a>(&'a Payload); + +impl Serialize for JsonPayload<'_> { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + match self.0 { + Payload::Json(value) => value.serialize(serializer), + Payload::Text(text) | Payload::Proto(text) => serializer.serialize_str(text), + Payload::Raw(bytes) => match serde_json::from_slice::<&RawValue>(bytes) { Review Comment: warning: sniffing Raw payloads as JSON breaks two ways. Pretty-printed JSON goes out verbatim through `RawValue`, newlines included, so one message spans several `.jsonl` lines; line readers get bad lines, and `[\n1,\n2\n]` even yields a bogus record `2`. `s3_sink` re-serializes compactly. It is also not reversible: Raw bytes `"AQI="` and Raw `[1, 2]` produce the same output. Fix: compact the sniffed JSON and tag which encoding was used, or always base64 Raw. ########## core/connectors/sinks/opendal_sink/src/formatter.rs: ########## @@ -0,0 +1,339 @@ +// 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. + +use std::collections::BTreeMap; + +use base64::Engine; +use chrono::{DateTime, Utc}; +use iggy_common::{HeaderKey, HeaderKind, HeaderValue}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata}; +use serde::{Serialize, Serializer, ser::SerializeMap}; +use serde_json::value::RawValue; + +use crate::OutputFormat; + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Formatter { + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, +} + +impl Formatter { + pub(crate) const fn new( + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, + ) -> Self { + Self { + output_format, + include_metadata, + include_headers, + } + } + + pub(crate) const fn output_format(&self) -> OutputFormat { + self.output_format + } + + pub(crate) fn format_batch( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<Vec<u8>, Error> { + let mut output = Vec::with_capacity(self.batch_capacity(messages, topic_metadata)); + if self.output_format == OutputFormat::JsonArray { + output.push(b'['); + } + + for (index, message) in messages.iter().enumerate() { + if self.output_format == OutputFormat::JsonArray && index > 0 { + output.push(b','); + } + + match self.output_format { + OutputFormat::JsonLines | OutputFormat::JsonArray => self.write_json_message( + &mut output, + message, + topic_metadata, + messages_metadata, + )?, + OutputFormat::Raw => self.write_raw_payload(&mut output, message)?, + } + + if self.output_format == OutputFormat::JsonLines { + output.push(b'\n'); + } + } + + if self.output_format == OutputFormat::JsonArray { + output.push(b']'); + } + Ok(output) + } + + fn write_json_message( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<(), Error> { + let timestamp = self + .include_metadata + .then(|| timestamp_to_rfc3339(message.timestamp)); + let json_message = JsonMessage { + offset: self.include_metadata.then_some(message.offset), + timestamp: timestamp.as_deref(), + stream: self + .include_metadata + .then_some(topic_metadata.stream.as_str()), + topic: self + .include_metadata + .then_some(topic_metadata.topic.as_str()), + partition_id: self + .include_metadata + .then_some(messages_metadata.partition_id), + headers: if self.include_headers { + message.headers.as_ref().map(JsonHeaders) + } else { + None + }, + payload: JsonPayload(&message.payload), + }; + + serde_json::to_writer(output, &json_message).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize message at offset {}: {error}", + message.offset + )) + }) + } + + fn write_raw_payload( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + ) -> Result<(), Error> { + match &message.payload { + Payload::Json(value) => serde_json::to_writer(output, value).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize raw payload at offset {}: {error}", + message.offset + )) + }), + Payload::Raw(bytes) | Payload::FlatBuffer(bytes) | Payload::Avro(bytes) => { + output.extend_from_slice(bytes); + Ok(()) + } + Payload::Text(text) | Payload::Proto(text) => { + output.extend_from_slice(text.as_bytes()); + Ok(()) + } + } + } + + fn batch_capacity( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + ) -> usize { + let metadata_size = if self.include_metadata { + topic_metadata + .stream + .len() + .saturating_add(topic_metadata.topic.len()) + .saturating_add(128) + } else if self.output_format == OutputFormat::Raw { + 0 + } else { + 16 + }; + + messages.iter().fold(2, |capacity, message| { + capacity + .saturating_add(payload_size_hint(&message.payload)) + .saturating_add(metadata_size) + }) + } +} + +#[derive(Serialize)] +struct JsonMessage<'a> { + #[serde(skip_serializing_if = "Option::is_none")] + offset: Option<u64>, + #[serde(skip_serializing_if = "Option::is_none")] + timestamp: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + stream: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + topic: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + partition_id: Option<u32>, + #[serde(skip_serializing_if = "Option::is_none")] + headers: Option<JsonHeaders<'a>>, + payload: JsonPayload<'a>, +} + +struct JsonHeaders<'a>(&'a BTreeMap<HeaderKey, HeaderValue>); + +impl Serialize for JsonHeaders<'_> { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + let mut map = serializer.serialize_map(Some(self.0.len()))?; + for (key, value) in self.0 { + serialize_header(&mut map, key.as_str().unwrap_or(""), value)?; + } + map.end() + } +} + +struct JsonPayload<'a>(&'a Payload); + +impl Serialize for JsonPayload<'_> { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + match self.0 { + Payload::Json(value) => value.serialize(serializer), + Payload::Text(text) | Payload::Proto(text) => serializer.serialize_str(text), Review Comment: warning: Proto text goes out as a JSON string. The connector-sink skill requires `json_document()` for it and `s3_sink` already does that, so a descriptor-less proto that holds JSON is double-encoded here. That also makes `README.md:128` ("Output formats match the S3 sink") untrue. Fix: route `Proto` through `json_document()` like `s3_sink`, falling back to the string only when it returns `None`. ########## core/connectors/sinks/opendal_sink/src/formatter.rs: ########## @@ -0,0 +1,339 @@ +// 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. + +use std::collections::BTreeMap; + +use base64::Engine; +use chrono::{DateTime, Utc}; +use iggy_common::{HeaderKey, HeaderKind, HeaderValue}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata}; +use serde::{Serialize, Serializer, ser::SerializeMap}; +use serde_json::value::RawValue; + +use crate::OutputFormat; + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Formatter { + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, +} + +impl Formatter { + pub(crate) const fn new( + output_format: OutputFormat, + include_metadata: bool, + include_headers: bool, + ) -> Self { + Self { + output_format, + include_metadata, + include_headers, + } + } + + pub(crate) const fn output_format(&self) -> OutputFormat { + self.output_format + } + + pub(crate) fn format_batch( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<Vec<u8>, Error> { + let mut output = Vec::with_capacity(self.batch_capacity(messages, topic_metadata)); + if self.output_format == OutputFormat::JsonArray { + output.push(b'['); + } + + for (index, message) in messages.iter().enumerate() { + if self.output_format == OutputFormat::JsonArray && index > 0 { + output.push(b','); + } + + match self.output_format { + OutputFormat::JsonLines | OutputFormat::JsonArray => self.write_json_message( + &mut output, + message, + topic_metadata, + messages_metadata, + )?, + OutputFormat::Raw => self.write_raw_payload(&mut output, message)?, + } + + if self.output_format == OutputFormat::JsonLines { + output.push(b'\n'); + } + } + + if self.output_format == OutputFormat::JsonArray { + output.push(b']'); + } + Ok(output) + } + + fn write_json_message( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + ) -> Result<(), Error> { + let timestamp = self + .include_metadata + .then(|| timestamp_to_rfc3339(message.timestamp)); + let json_message = JsonMessage { + offset: self.include_metadata.then_some(message.offset), + timestamp: timestamp.as_deref(), + stream: self + .include_metadata + .then_some(topic_metadata.stream.as_str()), + topic: self + .include_metadata + .then_some(topic_metadata.topic.as_str()), + partition_id: self + .include_metadata + .then_some(messages_metadata.partition_id), + headers: if self.include_headers { + message.headers.as_ref().map(JsonHeaders) + } else { + None + }, + payload: JsonPayload(&message.payload), + }; + + serde_json::to_writer(output, &json_message).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize message at offset {}: {error}", + message.offset + )) + }) + } + + fn write_raw_payload( + &self, + output: &mut Vec<u8>, + message: &ConsumedMessage, + ) -> Result<(), Error> { + match &message.payload { + Payload::Json(value) => serde_json::to_writer(output, value).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to serialize raw payload at offset {}: {error}", + message.offset + )) + }), + Payload::Raw(bytes) | Payload::FlatBuffer(bytes) | Payload::Avro(bytes) => { + output.extend_from_slice(bytes); + Ok(()) + } + Payload::Text(text) | Payload::Proto(text) => { + output.extend_from_slice(text.as_bytes()); + Ok(()) + } + } + } + + fn batch_capacity( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + ) -> usize { + let metadata_size = if self.include_metadata { + topic_metadata + .stream + .len() + .saturating_add(topic_metadata.topic.len()) + .saturating_add(128) + } else if self.output_format == OutputFormat::Raw { + 0 + } else { + 16 + }; + + messages.iter().fold(2, |capacity, message| { + capacity + .saturating_add(payload_size_hint(&message.payload)) + .saturating_add(metadata_size) + }) + } +} + +#[derive(Serialize)] +struct JsonMessage<'a> { + #[serde(skip_serializing_if = "Option::is_none")] + offset: Option<u64>, + #[serde(skip_serializing_if = "Option::is_none")] + timestamp: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + stream: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + topic: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + partition_id: Option<u32>, + #[serde(skip_serializing_if = "Option::is_none")] + headers: Option<JsonHeaders<'a>>, + payload: JsonPayload<'a>, +} + +struct JsonHeaders<'a>(&'a BTreeMap<HeaderKey, HeaderValue>); + +impl Serialize for JsonHeaders<'_> { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + let mut map = serializer.serialize_map(Some(self.0.len()))?; + for (key, value) in self.0 { + serialize_header(&mut map, key.as_str().unwrap_or(""), value)?; Review Comment: warning: `key.as_str().unwrap_or("")` turns every non-String header key into `""`, so two such headers produce duplicate JSON keys and readers keep only one of them. Fix: serialize non-String keys through `to_string_value()` (it exists on `HeaderField<T>`, so on `HeaderKey` too), and add a test with two numeric keys. ########## core/connectors/sinks/opendal_sink/config.toml: ########## @@ -0,0 +1,48 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +type = "sink" +key = "opendal" +enabled = true +version = 0 +name = "OpenDAL sink" +path = "../../target/release/libiggy_connector_opendal_sink" +plugin_config_format = "toml" +verbose = false +benchmark = false + +[[streams]] +stream = "example_stream" +topics = ["example_topic"] +schema = "json" +batch_length = 100 +poll_interval = "100ms" Review Comment: warning: master now rejects `poll_interval` on sink streams (`UnsupportedPollInterval`, #4629). After the rebase this example, the README config (`README.md:23`) and the integration fixture (`fixtures/opendal.rs:105`) fail to start. Git shows no conflict here because the files are new, so it is easy to miss. Fix: drop the key from all three. ########## core/integration/tests/connectors/opendal/opendal_sink.rs: ########## @@ -0,0 +1,76 @@ +// 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. + +use bytes::Bytes; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::{Identifier, MessageClient}; +use integration::{harness::seeds, iggy_harness}; + +use crate::connectors::fixtures::OpenDalSinkFixture; + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/opendal/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_json_batch_when_consumed_should_write_single_json_lines_object_to_fs( Review Comment: warning: the only e2e test uses `fs`, so S3, GCS and Azure never run through the dlopened plugin (reqwest, rustls and credential loading inside the cdylib). The repo already has a Floci S3 fixture (`fixtures/floci.rs`). Fix: add one S3 case through it, ideally including a failed write so the error path is covered end to end. ########## core/connectors/sinks/opendal_sink/README.md: ########## @@ -0,0 +1,165 @@ +# Apache Iggy OpenDAL Sink Connector + +Writes Iggy messages to storage services supported by Apache OpenDAL. + +## Configuration + +```toml +type = "sink" +key = "opendal" +enabled = true +version = 0 +name = "OpenDAL sink" +path = "../../target/release/libiggy_connector_opendal_sink" +plugin_config_format = "toml" +verbose = false +benchmark = false + +[[streams]] +stream = "events" +topics = ["orders"] +schema = "json" +batch_length = 100 +poll_interval = "100ms" +consumer_group = "opendal_sink" + +[plugin_config] +service = "fs" +path_prefix = "archive" +path_template = "{stream}/{topic}/{date}/{hour}" +output_format = "json_lines" +include_metadata = true +include_headers = false +max_attempts = 3 +retry_delay = "1s" +verbose_logging = false + +[plugin_config.options] +root = "/var/lib/iggy-objects" +``` + +### Plugin options + +| Option | Type | Default | Description | +| ------------------ | -------- | -------------------------------- | -------------------------------------------------------------------------------------- | +| `service` | String | **required** | OpenDAL service name | +| `path_prefix` | String | empty | Prefix before the rendered object path | +| `path_template` | String | `{stream}/{topic}/{date}/{hour}` | Object directory template | +| `options` | Map | empty | Secret OpenDAL options as quoted TOML strings; values of other types will fail to load | +| `output_format` | String | `json_lines` | `json_lines` (also accepts `jsonl` or `jsonlines`), `json_array`, or `raw` | +| `include_metadata` | Boolean | `true` | Add message metadata to JSON output | +| `include_headers` | Boolean | `false` | Add message headers to JSON output | +| `max_attempts` | Integer | `3` | Total write attempts (`max_retries` alias) | +| `retry_delay` | Duration | `1s` | Delay before the first retry | +| `verbose_logging` | Boolean | `false` | Log each consumed batch at info level | + +### Filesystem atomic writes + +For `service = "fs"`, the sink enables atomic writes by default. If +`plugin_config.options.atomic_write_dir` is not set, the sink uses +`<root>/.iggy-opendal-tmp`. Set `atomic_write_dir = ""` to disable atomic +writes, or set it to another directory on the same filesystem as `root`. + +### Default services + +The sink's default crate feature enables these OpenDAL services: + +- Azure Blob Storage +- Filesystem +- Google Cloud Storage +- Amazon S3 and compatible stores + +OpenDAL's in-memory service is always available and does not require a service feature. Review Comment: warning: `memory` accepts writes, so `service = "memory"` acks every batch into process RAM, grows without bound, and loses everything on restart, with offsets already committed. Fix: reject `memory` in `open()` and drop this line. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; Review Comment: warning: one consume batch becomes one PUT with no byte bound, up to `batch_length` x message size. S3 rejects a single PUT over 5 GiB and Azure over 5000 MiB as permanent errors, so the batch is dropped, and peak memory is about 3x the batch. On the open question from the earlier review: cap object size (default on, 8 MiB like `s3_sink`), split inside one `consume` with each chunk named by its own offset range, and keep not buffering across calls. The cap also lets the write timeout above have a sane default. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || operator.write(&path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL batch object '{path}' with offsets {}-{last_offset} \ + after {} attempt(s): {}", + first_message.offset, failure.attempts, failure.error + )) + })?; + + Ok(()) + } +} + +#[async_trait] +impl Sink for OpenDalSink { + async fn open(&mut self) -> Result<(), Error> { + let service = self.config.service.trim(); + if service.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL service cannot be empty".to_string(), + )); + } + if self.config.path_template.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL path_template cannot be empty".to_string(), + )); + } + + let output_format = OutputFormat::try_from(self.config.output_format.as_str())?; + let formatter = formatter::Formatter::new( + output_format, + self.config.include_metadata, + self.config.include_headers, + ); + opendal::install_default(); + let options = operator_options(service, &self.config.options); + let operator = Operator::via_iter(service, options).map_err(|error| { + Error::InitError(format!( + "Failed to create OpenDAL service '{service}': {error}" + )) + })?; + + if !operator.info().capability().write { + return Err(Error::InvalidConfigValue(format!( + "OpenDAL service '{service}' does not support writes" + ))); + } + + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, service: {service}, root: {}", + self.id, + operator.info().root() + ); + self.formatter = Some(formatter); + self.operator = Some(operator); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let Some(operator) = self.operator.as_ref() else { + return Err(Error::InitError( + "OpenDAL operator is not initialized".to_string(), + )); + }; + let Some(formatter) = self.formatter.as_ref() else { + return Err(Error::InitError( + "OpenDAL formatter is not initialized".to_string(), + )); + }; + if messages.is_empty() { + return Ok(()); + } + + if self.verbose { + info!( + "{CONNECTOR_NAME} connector ID: {} consuming {} messages, stream: {}, topic: {}, \ + partition_id: {}, current_offset: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset + ); + } + + let message_count = messages.len() as u64; + match self + .write_batch( + operator, + topic_metadata, + &messages_metadata, + &messages, + formatter, + ) + .await + { + Ok(()) => { + self.messages_processed + .fetch_add(message_count, Ordering::Relaxed); + debug!( + "{CONNECTOR_NAME} connector ID: {} uploaded one object with {} messages", + self.id, message_count + ); + Ok(()) + } + Err(write_error) => { + self.write_errors + .fetch_add(message_count, Ordering::Relaxed); + error!( + "{CONNECTOR_NAME} connector ID: {} failed to upload {} messages: {write_error}", + self.id, message_count + ); + Err(write_error) + } + } + } + + async fn close(&mut self) -> Result<(), Error> { + self.operator.take(); + self.formatter.take(); + info!( + "Closed {CONNECTOR_NAME} connector ID: {}, processed: {}, failed_messages: {}", + self.id, + self.messages_processed.load(Ordering::Relaxed), + self.write_errors.load(Ordering::Relaxed) + ); + Ok(()) + } +} + +fn default_path_template() -> String { + DEFAULT_PATH_TEMPLATE.to_string() +} + +fn default_output_format() -> String { + DEFAULT_OUTPUT_FORMAT.to_string() +} + +fn default_true() -> bool { + true +} + +fn operator_options( + service: &str, + options: &BTreeMap<String, SecretString>, +) -> Vec<(String, String)> { + let mut ret = Vec::with_capacity(options.len() + 1); + ret.extend( + options + .iter() + .map(|(key, value)| (key.clone(), value.expose_secret().to_owned())), + ); + + if service.eq_ignore_ascii_case("fs") Review Comment: warning: the atomic default only fires on an exact `root` key and the `fs` scheme name. OpenDAL lowercases option keys and accepts `file` as an fs alias, so `ROOT = ...`, `Root = ...` or `service = "file"` get in-place writes, and a crash leaves a torn object at the final key. A user `ATOMIC_WRITE_DIR` is missed by `contains_key` too, and the sink's own default then overrides it. Fix: compare keys case-insensitively and resolve the scheme the way OpenDAL does before deciding. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || operator.write(&path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL batch object '{path}' with offsets {}-{last_offset} \ + after {} attempt(s): {}", + first_message.offset, failure.attempts, failure.error + )) + })?; + + Ok(()) + } +} + +#[async_trait] +impl Sink for OpenDalSink { + async fn open(&mut self) -> Result<(), Error> { + let service = self.config.service.trim(); + if service.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL service cannot be empty".to_string(), + )); + } + if self.config.path_template.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL path_template cannot be empty".to_string(), + )); + } + + let output_format = OutputFormat::try_from(self.config.output_format.as_str())?; + let formatter = formatter::Formatter::new( + output_format, + self.config.include_metadata, + self.config.include_headers, + ); + opendal::install_default(); + let options = operator_options(service, &self.config.options); + let operator = Operator::via_iter(service, options).map_err(|error| { + Error::InitError(format!( + "Failed to create OpenDAL service '{service}': {error}" + )) + })?; + + if !operator.info().capability().write { Review Comment: warning: nothing checks that the target is writable. With a wrong bucket or bad credentials `open()` succeeds, every batch fails and is dropped (offsets are already committed), and the connector stays Running with only the error counter moving. Fix: write and delete a small sentinel object under the prefix in `open()` and fail on error. Unlike the removed `check()`, a write probe works with write-only credentials. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,752 @@ +// 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. + +use std::{ + collections::BTreeMap, + fmt, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, + retry::{RetryPolicy, parse_duration, retry_async}, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tracing::{debug, error, info}; + +use crate::path::{PathContext, object_path}; + +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_OUTPUT_FORMAT: &str = "json_lines"; +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Clone, Deserialize)] +pub struct OpenDalSinkConfig { + pub service: String, + #[serde(default)] + pub path_prefix: Option<String>, + #[serde(default = "default_path_template")] + pub path_template: String, + #[serde(default)] + pub options: BTreeMap<String, SecretString>, + #[serde(default = "default_output_format")] + pub output_format: String, + #[serde(default = "default_true")] + pub include_metadata: bool, + #[serde(default)] + pub include_headers: bool, + #[serde(default, alias = "max_retries")] + pub max_attempts: Option<u32>, + #[serde(default)] + pub retry_delay: Option<String>, + #[serde(default)] + pub verbose_logging: Option<bool>, +} + +impl fmt::Debug for OpenDalSinkConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("OpenDalSinkConfig") + .field("service", &self.service) + .field("path_prefix", &self.path_prefix) + .field("path_template", &self.path_template) + .field("option_keys", &self.options.keys().collect::<Vec<_>>()) + .field("output_format", &self.output_format) + .field("include_metadata", &self.include_metadata) + .field("include_headers", &self.include_headers) + .field("max_attempts", &self.max_attempts) + .field("retry_delay", &self.retry_delay) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + config: OpenDalSinkConfig, + path_prefix: String, + retry_policy: RetryPolicy, + verbose: bool, + operator: Option<Operator>, + formatter: Option<formatter::Formatter>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + let max_attempts = config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS); + + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + + let path_prefix = config + .path_prefix + .as_deref() + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let verbose = config.verbose_logging.unwrap_or(false); + + Self { + id, + config, + path_prefix, + retry_policy: RetryPolicy { + max_attempts, + base_delay: retry_delay, + max_delay: MAX_BACKOFF, + }, + verbose, + formatter: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( + &self, + operator: &Operator, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + formatter: &formatter::Formatter, + ) -> Result<(), Error> { + let Some(first_message) = messages.first() else { + return Ok(()); + }; + let last_offset = messages + .last() + .map_or(first_message.offset, |message| message.offset); + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: first_message.timestamp, + }; + let path = object_path( + &self.path_prefix, + &self.config.path_template, + &context, + first_message.offset, + last_offset, + formatter.output_format(), + )?; + let data = formatter.format_batch(messages, topic_metadata, messages_metadata)?; + let buffer = Buffer::from(data); + let retry_context = format!("{CONNECTOR_NAME} connector ID {} batch write", self.id); + + retry_async( + self.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || operator.write(&path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL batch object '{path}' with offsets {}-{last_offset} \ + after {} attempt(s): {}", + first_message.offset, failure.attempts, failure.error + )) + })?; + + Ok(()) + } +} + +#[async_trait] +impl Sink for OpenDalSink { + async fn open(&mut self) -> Result<(), Error> { + let service = self.config.service.trim(); + if service.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL service cannot be empty".to_string(), + )); + } + if self.config.path_template.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL path_template cannot be empty".to_string(), + )); + } + + let output_format = OutputFormat::try_from(self.config.output_format.as_str())?; + let formatter = formatter::Formatter::new( + output_format, + self.config.include_metadata, + self.config.include_headers, + ); + opendal::install_default(); + let options = operator_options(service, &self.config.options); + let operator = Operator::via_iter(service, options).map_err(|error| { + Error::InitError(format!( + "Failed to create OpenDAL service '{service}': {error}" + )) + })?; + + if !operator.info().capability().write { + return Err(Error::InvalidConfigValue(format!( + "OpenDAL service '{service}' does not support writes" + ))); + } + + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, service: {service}, root: {}", + self.id, + operator.info().root() + ); + self.formatter = Some(formatter); + self.operator = Some(operator); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let Some(operator) = self.operator.as_ref() else { + return Err(Error::InitError( + "OpenDAL operator is not initialized".to_string(), + )); + }; + let Some(formatter) = self.formatter.as_ref() else { + return Err(Error::InitError( + "OpenDAL formatter is not initialized".to_string(), + )); + }; + if messages.is_empty() { + return Ok(()); + } + + if self.verbose { + info!( + "{CONNECTOR_NAME} connector ID: {} consuming {} messages, stream: {}, topic: {}, \ + partition_id: {}, current_offset: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset + ); + } + + let message_count = messages.len() as u64; + match self + .write_batch( + operator, + topic_metadata, + &messages_metadata, + &messages, + formatter, + ) + .await + { + Ok(()) => { + self.messages_processed + .fetch_add(message_count, Ordering::Relaxed); + debug!( + "{CONNECTOR_NAME} connector ID: {} uploaded one object with {} messages", + self.id, message_count + ); + Ok(()) + } + Err(write_error) => { + self.write_errors + .fetch_add(message_count, Ordering::Relaxed); + error!( + "{CONNECTOR_NAME} connector ID: {} failed to upload {} messages: {write_error}", + self.id, message_count + ); + Err(write_error) + } + } + } + + async fn close(&mut self) -> Result<(), Error> { + self.operator.take(); + self.formatter.take(); + info!( + "Closed {CONNECTOR_NAME} connector ID: {}, processed: {}, failed_messages: {}", + self.id, + self.messages_processed.load(Ordering::Relaxed), + self.write_errors.load(Ordering::Relaxed) + ); + Ok(()) + } +} + +fn default_path_template() -> String { + DEFAULT_PATH_TEMPLATE.to_string() +} + +fn default_output_format() -> String { + DEFAULT_OUTPUT_FORMAT.to_string() +} + +fn default_true() -> bool { + true +} + +fn operator_options( + service: &str, + options: &BTreeMap<String, SecretString>, +) -> Vec<(String, String)> { + let mut ret = Vec::with_capacity(options.len() + 1); + ret.extend( + options + .iter() + .map(|(key, value)| (key.clone(), value.expose_secret().to_owned())), + ); + + if service.eq_ignore_ascii_case("fs") + && !options.contains_key("atomic_write_dir") + && let Some(root) = options.get("root") + { + let atomic_write_dir = Path::new(root.expose_secret()) Review Comment: warning: the fs backend fsyncs the temp file and renames it, but never fsyncs the parent directory. A power loss after a successful `consume` can lose the rename, so the acked object disappears while its offset stays committed. Fix: fsync the target directory after the write for `fs`, or document that fs durability stops at the rename. ########## core/connectors/sinks/opendal_sink/src/path.rs: ########## @@ -0,0 +1,162 @@ +// 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. + +use chrono::{DateTime, Utc}; +use iggy_connector_sdk::Error; + +use crate::OutputFormat; + +pub(crate) struct PathContext<'a> { + pub(crate) stream: &'a str, + pub(crate) topic: &'a str, + pub(crate) partition_id: u32, + pub(crate) first_timestamp_micros: u64, +} + +pub(crate) fn object_path( + path_prefix: &str, + path_template: &str, + context: &PathContext<'_>, + offset_start: u64, + offset_end: u64, + format: OutputFormat, +) -> Result<String, Error> { + let rendered = render_template(path_template, context)?; + + // Partition ID is always embedded in the filename to prevent cross-partition + // key collisions because partitions have independent offset spaces starting at + // 0. + let filename = format!( Review Comment: warning: the key is unique only if the template has both `{stream}` and `{topic}`, and every write replaces what is already there. A multi-topic sink with a `{date}/{hour}` template overwrites already-acked objects of another topic. A deleted and recreated topic restarts at offset 0 and overwrites the old archive; the fixture's `{stream}/{topic}` template hits this. Fix: reject templates without `{stream}` and `{topic}` at open. `if_not_exists` would cover recreate, but make it opt-in or probe it at open: S3-compatible stores without If-None-Match would fail every write. -- 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]
