hubcio commented on code in PR #4123: URL: https://github.com/apache/iggy/pull/4123#discussion_r4070691187
########## core/connectors/sinks/opendal_sink/src/path.rs: ########## @@ -0,0 +1,150 @@ +// 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!( + "{:05}-{:020}-{:020}.{}", + context.partition_id, + offset_start, + offset_end, + format.file_extension() + ); + + if path_prefix.is_empty() { + Ok(format!("{rendered}/{filename}")) + } else { + Ok(format!("{path_prefix}/{rendered}/{filename}")) + } +} + +fn render_template(template: &str, context: &PathContext<'_>) -> Result<String, Error> { + let timestamp = timestamp_to_datetime(context.first_timestamp_micros)?; + let date = timestamp.format("%Y-%m-%d").to_string(); + let hour = timestamp.format("%H").to_string(); + let timestamp_millis = (context.first_timestamp_micros / 1_000).to_string(); + + Ok(template + .replace("{stream}", &sanitize_path_segment(context.stream)) + .replace("{topic}", &sanitize_path_segment(context.topic)) + .replace("{partition}", &context.partition_id.to_string()) + .replace("{date}", &date) + .replace("{hour}", &hour) + .replace("{timestamp}", ×tamp_millis)) +} + +fn sanitize_path_segment(segment: &str) -> String { Review Comment: warning: `a b` and `a_b` both sanitize to `a_b`, so the later write hides the earlier topic's object, and `..` passes too and always fails. percent-encode the segment or hash the raw name. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: 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], + output_format: OutputFormat, + ) -> 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, + output_format, + )?; + let entries = messages + .iter() + .map(|message| { + formatter::format_message( + message, + topic_metadata, + messages_metadata, + self.config.include_metadata, + self.config.include_headers, + output_format, + ) + }) + .collect::<Result<Vec<_>, _>>()?; + let data = formatter::finalize_buffer(entries.iter().map(Vec::as_slice), output_format); + 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: warning: with the shipped `fs` config opendal truncates the final key and writes in place, so a reader can see a partial object and a crash leaves one. set `atomic_write_dir`, and document that fs writes are not atomic. ########## core/connectors/sinks/opendal_sink/README.md: ########## @@ -0,0 +1,161 @@ +# 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 service options | +| `output_format` | String | `json_lines` | `json_lines`, `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 | +| `retry_delay` | Duration | `1s` | Delay before the first retry | +| `verbose_logging` | Boolean | `false` | Log each consumed batch at info level | + +### 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. + +Use `--no-default-features` when building to disable the four default services. + +### Additional services + +When building the sink directly, enable another OpenDAL service through Cargo's feature syntax. For example, this command builds the sink with CompFS and without the default services: + +```bash +cargo build --release -p iggy_connector_opendal_sink \ + --no-default-features \ + --features opendal/services-compfs +``` + +When using the sink as a dependency, add `opendal` as a direct dependency and enable additional features: + +```toml +iggy_connector_opendal_sink = { version = "0.5.0-edge.4", default-features = false } Review Comment: warning: this dependency cannot resolve - the crate sets `publish = false`, so nothing is on a registry. replace the snippet with a `cargo build` command. ########## core/integration/tests/connectors/fixtures/opendal.rs: ########## @@ -0,0 +1,161 @@ +// 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::HashMap, + fs, + io::ErrorKind, + path::{Path, PathBuf}, + time::Duration, +}; + +use async_trait::async_trait; +use integration::harness::{TestBinaryError, TestFixture}; +use tempfile::TempDir; +use tokio::time::sleep; + +const CONFIG_DIR_ENV: &str = "IGGY_CONNECTORS_CONNECTORS_CONFIG_DIR"; +const POLL_ATTEMPTS: usize = 100; +const POLL_INTERVAL: Duration = Duration::from_millis(50); + +pub struct OpenDalSinkFixture { + _temp_dir: TempDir, + object_root: PathBuf, + config_dir: PathBuf, +} + +impl OpenDalSinkFixture { + pub async fn wait_for_object( + &self, + offset_start: u64, + offset_end: u64, + ) -> Result<Vec<u8>, TestBinaryError> { + let path = self.object_path(offset_start, offset_end); + + for _ in 0..POLL_ATTEMPTS { + match tokio::fs::read(&path).await { + Ok(payload) => return Ok(payload), Review Comment: warning: returns on the first successful read, but the sink writes in place, so a poll that lands mid-write reads an empty object and the assertion fires instead of retrying. poll on the payload comparison. ########## core/connectors/sinks/opendal_sink/README.md: ########## @@ -0,0 +1,161 @@ +# 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 service options | Review Comment: nit: values must be quoted strings in TOML, since `SecretString` deserializes through `String`, so a bare number or bool fails at load. say so in the row. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + pub max_attempts: Option<u32>, Review Comment: nit: `s3_sink` accepts both `max_attempts` and `max_retries`, and five other sinks use `max_retries`. add `#[serde(alias = "max_retries")]` so a copied config is not silently ignored. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: 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], + output_format: OutputFormat, + ) -> 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, + output_format, + )?; + let entries = messages Review Comment: warning: collects a `Vec<u8>` per message, then `finalize_buffer` copies all of them into a second buffer. format straight into one buffer, and move the `JsonArray` bracket logic there first. ########## core/connectors/sinks/opendal_sink/src/formatter.rs: ########## @@ -0,0 +1,215 @@ +// 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 base64::Engine; +use chrono::{DateTime, Utc}; +use iggy_common::HeaderKind; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata, owned_value_to_serde_json, +}; +use serde::Serialize; +use serde_json::{Map, Value}; + +use crate::OutputFormat; + +#[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<Value>, + payload: Value, +} + +pub(crate) fn format_message( + message: &ConsumedMessage, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + include_metadata: bool, + include_headers: bool, + format: OutputFormat, +) -> Result<Vec<u8>, Error> { + match format { + OutputFormat::JsonLines | OutputFormat::JsonArray => format_json( + message, + topic_metadata, + messages_metadata, + include_metadata, + include_headers, + format, + ), + OutputFormat::Raw => message.payload.try_to_bytes().map_err(|error| { + Error::CannotStoreData(format!( + "Failed to extract raw bytes at offset {}: {error}", + message.offset + )) + }), + } +} + +fn format_json( + message: &ConsumedMessage, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + include_metadata: bool, + include_headers: bool, + format: OutputFormat, +) -> Result<Vec<u8>, Error> { + let timestamp = if include_metadata { + Some(timestamp_to_rfc3339(message.timestamp)) + } else { + None + }; + + let json_message = JsonMessage { + offset: if include_metadata { + Some(message.offset) + } else { + None + }, + timestamp: timestamp.as_deref(), + stream: if include_metadata { + Some(&topic_metadata.stream) + } else { + None + }, + topic: if include_metadata { + Some(&topic_metadata.topic) + } else { + None + }, + partition_id: if include_metadata { + Some(messages_metadata.partition_id) + } else { + None + }, + headers: if include_headers { + message.headers.as_ref().map(serialize_headers) + } else { + None + }, + payload: payload_to_json_value(&message.payload), Review Comment: warning: rebuilds the payload tree before serializing, which allocates a node per element and a `String` per key. serialize in place with `serde_json::to_writer` and a wrapper that maps the six `Payload` variants. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: None, + operator: None, + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } + + async fn write_batch( Review Comment: warning: one object per poll batch means nothing bounds object size in bytes. `s3_sink` rotates at 8 MiB, but the connector guidance says not to buffer across `consume` - which rule wins here? ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: 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], + output_format: OutputFormat, + ) -> 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, + output_format, + )?; + let entries = messages + .iter() + .map(|message| { + formatter::format_message( + message, + topic_metadata, + messages_metadata, + self.config.include_metadata, + self.config.include_headers, + output_format, + ) + }) + .collect::<Result<Vec<_>, _>>()?; + let data = formatter::finalize_buffer(entries.iter().map(Vec::as_slice), output_format); + 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())?; + opendal::install_default(); + let options = self + .config + .options + .iter() + .map(|(key, value)| (key.clone(), value.expose_secret().to_owned())); + 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" + ))); + } + + operator.check().await.map_err(|error| { Review Comment: warning: `operator.check()` sends a list request, so a write-only credential fails here and the connector never starts. is listing a documented requirement, or does this need a write probe? ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: 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], + output_format: OutputFormat, + ) -> 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, + output_format, + )?; + let entries = messages + .iter() + .map(|message| { + formatter::format_message( + message, + topic_metadata, + messages_metadata, + self.config.include_metadata, + self.config.include_headers, + output_format, + ) + }) + .collect::<Result<Vec<_>, _>>()?; + let data = formatter::finalize_buffer(entries.iter().map(Vec::as_slice), output_format); + 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())?; + opendal::install_default(); + let options = self + .config + .options + .iter() + .map(|(key, value)| (key.clone(), value.expose_secret().to_owned())); + 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" + ))); + } + + operator.check().await.map_err(|error| { + Error::InitError(format!( + "OpenDAL service '{service}' connectivity check failed: {error}" + )) + })?; + + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, service: {service}, root: {}", + self.id, + operator.info().root() + ); + self.output_format = Some(output_format); + 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(output_format) = self.output_format else { + return Err(Error::InitError( + "OpenDAL output format 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, + output_format, + ) + .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.output_format.take(); + info!( + "Closed {CONNECTOR_NAME} connector ID: {}, processed: {}, errors: {}", + 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 +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum OutputFormat { + JsonLines, + JsonArray, + Raw, +} + +impl TryFrom<&str> for OutputFormat { + type Error = Error; + + fn try_from(value: &str) -> Result<Self, Self::Error> { + match value.to_lowercase().as_str() { + "json_lines" | "jsonl" | "jsonlines" => Ok(Self::JsonLines), Review Comment: nit: `jsonl` and `jsonlines` are accepted here but the README lists only three names. add them to the table, since `s3_sink` documents the same pair. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,759 @@ +// 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, + 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)] + 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>, + output_format: Option<OutputFormat>, + 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, + output_format: 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], + output_format: OutputFormat, + ) -> 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, + output_format, + )?; + let entries = messages + .iter() + .map(|message| { + formatter::format_message( + message, + topic_metadata, + messages_metadata, + self.config.include_metadata, + self.config.include_headers, + output_format, + ) + }) + .collect::<Result<Vec<_>, _>>()?; + let data = formatter::finalize_buffer(entries.iter().map(Vec::as_slice), output_format); + 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())?; + opendal::install_default(); + let options = self + .config + .options + .iter() + .map(|(key, value)| (key.clone(), value.expose_secret().to_owned())); + 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" + ))); + } + + operator.check().await.map_err(|error| { + Error::InitError(format!( + "OpenDAL service '{service}' connectivity check failed: {error}" + )) + })?; + + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, service: {service}, root: {}", + self.id, + operator.info().root() + ); + self.output_format = Some(output_format); + 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(output_format) = self.output_format else { + return Err(Error::InitError( + "OpenDAL output format 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, + output_format, + ) + .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); Review Comment: nit: `write_errors` counts messages, but `close()` logs it under an errors label. rename the label - counting batches instead breaks the test at line 609. ########## core/connectors/sinks/opendal_sink/Cargo.toml: ########## @@ -0,0 +1,68 @@ +# 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. + +[package] +name = "iggy_connector_opendal_sink" +version = "0.5.0-edge.4" Review Comment: nit: `0.5.0-edge.4` is the only prerelease among the connector plugins, and `bump-version.sh` does not list this crate, so a release bump skips it. use `0.5.0` and add the crate to `CONNECTOR_SINK_COMPONENTS`. -- 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]
