spetz commented on code in PR #4123: URL: https://github.com/apache/iggy/pull/4123#discussion_r4440811224
########## core/connectors/sinks/opendal_sink/src/formatter.rs: ########## @@ -0,0 +1,957 @@ +// 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::{ + borrow::Cow, + collections::{BTreeMap, HashSet}, + io::{self, Write}, +}; + +use base64::{display::Base64Display, engine::general_purpose::STANDARD}; +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 crate::config::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 + } + + /// Returns the serialized object bytes and the number of input messages consumed. + pub(crate) fn format_chunk( + &self, + messages: &[ConsumedMessage], + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + max_bytes: usize, + ) -> Result<(Vec<u8>, usize), Error> { + if messages.is_empty() { + return Ok((Vec::new(), 0)); + } + + let mut output = BoundedWriter { + bytes: Vec::with_capacity(self.chunk_capacity(messages, topic_metadata, max_bytes)), + limit: if self.output_format == OutputFormat::JsonArray { + max_bytes.saturating_sub(1) + } else { + max_bytes + }, + exceeded: false, + }; + if self.output_format == OutputFormat::JsonArray { + output + .write_all(b"[") + .map_err(|_| oversized_message(&messages[0], max_bytes))?; + } + + let mut consumed = 0; + for message in messages { + let checkpoint = output.bytes.len(); + let result = (|| { + if self.output_format == OutputFormat::JsonArray && consumed > 0 { + output.write_all(b",").map_err(serde_json::Error::io)?; + } + + 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.write_all(b"\n").map_err(serde_json::Error::io)?; + } + Ok::<(), serde_json::Error>(()) + })(); + + if let Err(error) = result { + output.bytes.truncate(checkpoint); + if consumed > 0 { + break; + } + return Err(if output.exceeded { Review Comment: A message larger than `max_file_size` is rejected and counted as failed, and because the runtime commits offsets on poll it is gone for good. The default cap is 8 MiB while the server accepts payloads up to 64 MB (`core/common/src/types/message/iggy_message.rs:44`), so every message between the two is dropped with the shipped config, and with `json_lines` a Raw payload above about 6 MiB is enough because of Base64. `given_oversized_message_when_consuming_should_keep_other_messages` pins that behavior. The cap exists to keep one PUT under the 5 GiB and 5000 MiB request limits, which a single message can never reach, and `s3_sink` appends before it checks its 8 MiB rotation (`core/connectors/sinks/s3_sink/src/sink.rs:270`), so it writes such a message in an object above the cap instead of losing it. Because `consume` then returns `Err` for the whole run, the runtime adds none of the stored messages to `messages_processed` (`core/connectors/runtime/src/sink.rs:1011`), so one oversize d message in a run of 1000 shows up as 0 processed and 1 error. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,1066 @@ +// 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::{ + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, retry::retry_async, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +use crate::{ + config::{OpenDalSinkConfig, ValidatedConfig, Validation}, + path::{PathContext, object_path}, +}; + +mod config; +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const ABORT_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + state: Option<SinkState>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +#[derive(Debug)] +enum SinkState { + Config(OpenDalSinkConfig), + Running(RunningSink), +} + +#[derive(Debug)] +struct RunningSink { + config: ValidatedConfig, + operator: Operator, + formatter: formatter::Formatter, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + Self { + id, + state: Some(SinkState::Config(config)), + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } +} + +impl RunningSink { + async fn write_object(&self, id: u32, path: &str, buffer: Buffer) -> Result<(), Error> { + let retry_context = format!("{CONNECTOR_NAME} connector ID {id} object write"); + retry_async( + self.config.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || self.write_attempt(id, path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL object '{path}' after {} attempt(s): {}", + failure.attempts, failure.error + )) + }) + } + + async fn write_attempt(&self, id: u32, path: &str, buffer: Buffer) -> opendal::Result<()> { + let duration = self.write_timeout_for(buffer.len()); + let deadline = Instant::now() + duration; + let mut writer = timeout_at( + deadline, + self.operator + .writer_with(path) + .if_not_exists(self.config.if_not_exists), + ) + .await + .map_err(|_| write_timeout_error())??; + let result = timeout_at(deadline, async { + writer.write(buffer).await?; + writer.close().await?; + Ok(()) + }) + .await + .unwrap_or_else(|_| Err(write_timeout_error())); + + if result.is_err() { + match timeout(ABORT_TIMEOUT, writer.abort()).await { + Ok(Ok(())) => {} + Ok(Err(abort_error)) => warn!( + "{CONNECTOR_NAME} connector ID: {id} could not abort object '{path}': {abort_error}" + ), + Err(_) => { + warn!("{CONNECTOR_NAME} connector ID: {id} timed out aborting object '{path}'") + } + } + } + result + } + + /// Adds one second per MiB to the base timeout, scaled to the buffer size. + fn write_timeout_for(&self, buffer_size: usize) -> Duration { + self.config + .write_timeout + .saturating_add(Duration::from_micros( + (buffer_size as u64).saturating_mul(1_000_000) / (1024 * 1024), + )) + } + + async fn check_access(&self, id: u32) -> Result<(), Error> { + let probe_name = format!(".iggy-write-probe-{}", Uuid::new_v4()); + let path_prefix = &self.config.path_prefix; + let path = if path_prefix.is_empty() { + probe_name + } else { + format!("{path_prefix}/{probe_name}") + }; + let write_result = self.write_object(id, &path, Buffer::from("iggy")).await; + let delete_result = timeout(self.config.write_timeout, self.operator.delete(&path)).await; Review Comment: The probe needs more than the sink uses and proves less than the README says. It fails `open` when it cannot delete what it wrote, and `build_operator` rejects any service without delete capability (`core/connectors/sinks/opendal_sink/src/config.rs:302`), so a credential with `s3:PutObject` and no `s3:DeleteObject` still never starts, which is the write-only setup the earlier thread on `check()` asked to keep working. `s3_sink` probes with a multipart initiate and abort that needs no delete and publishes nothing (`core/connectors/sinks/s3_sink/src/sink.rs:189`). The delete also runs once with no retry while the write gets up to three attempts, so one transient 503 on it fails `open` even with full permissions, and the runtime does not retry a failed open. With `if_not_exists = true` the probe writes a fresh random key, which succeeds whether or not the store honors `If-None-Match`. OpenDAL reports `write_with_if_not_exists` for every `s3` endpoint (`opendal-service-s3` 0.59.1, `ba ckend.rs:1080`), so a store without conditional writes passes and then replaces existing objects, against the claim at `core/connectors/sinks/opendal_sink/README.md:192`. ########## core/connectors/sinks/opendal_sink/src/config.rs: ########## @@ -0,0 +1,617 @@ +// 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. + +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; +use std::{collections::BTreeMap, fmt, path::Path, time::Duration}; + +use iggy_connector_sdk::{ + Error, + retry::{RetryPolicy, parse_duration}, +}; +use opendal::Operator; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tokio::time::timeout; + +use crate::formatter::Formatter; + +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); +const DEFAULT_MAX_FILE_SIZE: &str = "8MiB"; +const DEFAULT_WRITE_TIMEOUT: &str = "60s"; +const MAX_OBJECT_SIZE: u64 = 5000 * 1024 * 1024; + +#[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 max_file_size: Option<String>, + #[serde(default)] + pub write_timeout: Option<String>, + #[serde(default)] + pub if_not_exists: Option<bool>, + #[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("max_file_size", &self.max_file_size) + .field("write_timeout", &self.write_timeout) + .field("if_not_exists", &self.if_not_exists) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub(crate) struct ValidatedConfig { + pub(crate) service: String, + pub(crate) path_prefix: String, + pub(crate) path_template: String, + pub(crate) options: BTreeMap<String, SecretString>, + pub(crate) max_file_size: usize, + pub(crate) output_format: OutputFormat, + pub(crate) include_metadata: bool, + pub(crate) include_headers: bool, + pub(crate) retry_policy: RetryPolicy, + pub(crate) write_timeout: Duration, + pub(crate) if_not_exists: bool, + pub(crate) verbose_logging: bool, +} + +#[derive(Debug)] +pub(crate) enum Validation { + Valid(ValidatedConfig), + Invalid { + config: OpenDalSinkConfig, + error: Error, + }, +} + +impl OpenDalSinkConfig { + pub(crate) fn validate(self) -> Validation { + let checked = (|| { + let mut service = self.service.trim().to_ascii_lowercase(); Review Comment: `Operator::via_iter` parses any `service` that contains `://` as a URI (`opendal-core` 0.59.1, `types/operator/uri.rs:52`) and `FsConfig::from_uri` takes the root from its path, but these checks compare the whole string with `fs` and `memory`. `service = "fs:///var/lib/iggy"` therefore skips the root check, the default `atomic_write_dir` and the same-device check and writes in place at the final key, and `service = "memory://"` gets past the memory ban. ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,1066 @@ +// 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::{ + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, retry::retry_async, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +use crate::{ + config::{OpenDalSinkConfig, ValidatedConfig, Validation}, + path::{PathContext, object_path}, +}; + +mod config; +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const ABORT_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + state: Option<SinkState>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +#[derive(Debug)] +enum SinkState { + Config(OpenDalSinkConfig), + Running(RunningSink), +} + +#[derive(Debug)] +struct RunningSink { + config: ValidatedConfig, + operator: Operator, + formatter: formatter::Formatter, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + Self { + id, + state: Some(SinkState::Config(config)), + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } +} + +impl RunningSink { + async fn write_object(&self, id: u32, path: &str, buffer: Buffer) -> Result<(), Error> { + let retry_context = format!("{CONNECTOR_NAME} connector ID {id} object write"); + retry_async( + self.config.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || self.write_attempt(id, path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL object '{path}' after {} attempt(s): {}", + failure.attempts, failure.error + )) + }) + } + + async fn write_attempt(&self, id: u32, path: &str, buffer: Buffer) -> opendal::Result<()> { + let duration = self.write_timeout_for(buffer.len()); + let deadline = Instant::now() + duration; + let mut writer = timeout_at( + deadline, + self.operator + .writer_with(path) + .if_not_exists(self.config.if_not_exists), + ) + .await + .map_err(|_| write_timeout_error())??; + let result = timeout_at(deadline, async { + writer.write(buffer).await?; + writer.close().await?; + Ok(()) + }) + .await + .unwrap_or_else(|_| Err(write_timeout_error())); + + if result.is_err() { + match timeout(ABORT_TIMEOUT, writer.abort()).await { + Ok(Ok(())) => {} + Ok(Err(abort_error)) => warn!( + "{CONNECTOR_NAME} connector ID: {id} could not abort object '{path}': {abort_error}" + ), + Err(_) => { + warn!("{CONNECTOR_NAME} connector ID: {id} timed out aborting object '{path}'") + } + } + } + result + } + + /// Adds one second per MiB to the base timeout, scaled to the buffer size. + fn write_timeout_for(&self, buffer_size: usize) -> Duration { + self.config + .write_timeout + .saturating_add(Duration::from_micros( + (buffer_size as u64).saturating_mul(1_000_000) / (1024 * 1024), + )) + } + + async fn check_access(&self, id: u32) -> Result<(), Error> { + let probe_name = format!(".iggy-write-probe-{}", Uuid::new_v4()); + let path_prefix = &self.config.path_prefix; + let path = if path_prefix.is_empty() { + probe_name + } else { + format!("{path_prefix}/{probe_name}") + }; + let write_result = self.write_object(id, &path, Buffer::from("iggy")).await; + let delete_result = timeout(self.config.write_timeout, self.operator.delete(&path)).await; + write_result + .map_err(|error| Error::InitError(format!("OpenDAL write probe failed: {error}")))?; + match delete_result { + Ok(Ok(())) => Ok(()), + Ok(Err(error)) => Err(Error::InitError(format!( + "OpenDAL write probe could not be deleted: {error}" + ))), + Err(_) => Err(Error::InitError( + "OpenDAL write probe deletion timed out".to_string(), + )), + } + } +} + +#[async_trait] +impl Sink for OpenDalSink { + async fn open(&mut self) -> Result<(), Error> { + let config = match self.state.take() { + Some(SinkState::Config(config)) => config, + Some(running @ SinkState::Running(_)) => { + self.state = Some(running); + return Ok(()); + } + None => { + return Err(Error::InitError( + "OpenDAL sink was closed or failed to initialize".to_string(), + )); + } + }; + let config = match config.validate() { + Validation::Valid(config) => config, + Validation::Invalid { config, error } => { + self.state = Some(SinkState::Config(config)); + return Err(error); + } + }; + let formatter = config.build_formatter(); + let operator = config.build_operator().await?; + let sink = RunningSink { + config, + operator, + formatter, + }; + sink.check_access(self.id).await?; + + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, service: {}, root: {}", + self.id, + sink.config.service, + sink.operator.info().root() + ); + self.state = Some(SinkState::Running(sink)); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let Some(SinkState::Running(state)) = &self.state else { + return Err(Error::InitError( + "OpenDAL sink is not initialized".to_string(), + )); + }; + let formatter = &state.formatter; + if messages.is_empty() { + return Ok(()); + } + + if state.config.verbose_logging { + 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 mut remaining = messages.as_slice(); + let mut last_error = None; + while !remaining.is_empty() { + let (data, count) = match formatter.format_chunk( + remaining, + topic_metadata, + &messages_metadata, + state.config.max_file_size, + ) { + Ok(chunk) => chunk, + Err(format_error) => { + self.write_errors.fetch_add(1, Ordering::Relaxed); + error!( + "{CONNECTOR_NAME} connector ID: {} cannot format message at offset {}: {format_error}", + self.id, remaining[0].offset + ); + last_error = Some(format_error); + remaining = &remaining[1..]; + continue; + } + }; + let chunk = &remaining[..count]; + let context = PathContext { + stream: &topic_metadata.stream, + topic: &topic_metadata.topic, + partition_id: messages_metadata.partition_id, + first_timestamp_micros: chunk[0].timestamp, + }; + let result = async { + let path = object_path( + &state.config.path_prefix, + &state.config.path_template, + &context, + chunk[0].offset, + chunk[count - 1].offset, + formatter.output_format(), + )?; + state.write_object(self.id, &path, Buffer::from(data)).await + } + .await; + let message_count = count as u64; + match result { + Ok(()) => { + self.messages_processed + .fetch_add(message_count, Ordering::Relaxed); + debug!( + "{CONNECTOR_NAME} connector ID: {} uploaded one object with {} messages", + self.id, message_count + ); + } + Err(write_error) => { Review Comment: Once a chunk has used all its attempts on timeouts, the loop moves to the next chunk against the same backend and spends the full budget again, so how long one `consume` can block grows with the batch rather than with `write_timeout`. With the defaults a chunk costs up to 3 attempts of 68 s, and the runtime's default `batch_length` of 1000 (`core/connectors/runtime/src/sink.rs:52`) with 64 KiB messages makes 8 chunks, about 28 minutes against a backend that accepts the connection and never answers. The runtime calls `consume` inline on a tokio worker (`core/connectors/runtime/src/sink.rs:1002`) and the SDK holds the `INSTANCES` entry across `block_on` (`core/connectors/sdk/src/sink.rs:299`), so when `stop_connector` gives up on its short task budget and calls `iggy_sink_close`, `INSTANCES.remove` (`core/connectors/sdk/src/sink.rs:318`) waits for that entry and stop or SIGTERM hangs for the whole time. Each stalled call also pins its runtime worker thread, so as many stalled topics as worker threads stop every other connector and the HTTP API with them. ########## core/integration/tests/connectors/fixtures/opendal.rs: ########## @@ -0,0 +1,382 @@ +// 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}, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use async_trait::async_trait; +use integration::harness::{TestBinaryError, TestFixture}; +use s3::Bucket; +use tempfile::TempDir; +use tokio::time::sleep; +use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate, matchers::method}; + +use crate::connectors::fixtures::{ + self, S3SinkOps, + floci::{self, ACCESS_KEY, FlociContainer, REGION, SECRET_KEY}, +}; + +const CONFIG_DIR_ENV: &str = "IGGY_CONNECTORS_CONNECTORS_CONFIG_DIR"; +const TEST_BUCKET: &str = "iggy-opendal-test"; +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, + expected: &[u8], + ) -> Result<(), TestBinaryError> { + let path = self.object_path(offset_start, offset_end); + + for _ in 0..POLL_ATTEMPTS { + match tokio::fs::read(&path).await { + Ok(payload) if payload == expected => return Ok(()), + Ok(_) => sleep(POLL_INTERVAL).await, + Err(error) if error.kind() == ErrorKind::NotFound => sleep(POLL_INTERVAL).await, + Err(source) => return Err(TestBinaryError::FileSystemError { path, source }), + } + } + + Err(TestBinaryError::InvalidState { + message: format!( + "OpenDAL object did not contain the expected payload: {}", + path.display() + ), + }) + } + + fn object_path(&self, offset_start: u64, offset_end: u64) -> PathBuf { + self.object_root.join(format!( + "iggy/messages/test_stream/test_topic/\ + 00000-{offset_start:020}-{offset_end:020}.jsonl" + )) + } +} + +#[async_trait] +impl TestFixture for OpenDalSinkFixture { + async fn setup() -> Result<Self, TestBinaryError> { + let temp_dir = TempDir::new().map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "OpenDalSinkFixture".to_string(), + message: format!("Failed to create temporary directory: {error}"), + })?; + let config_dir = temp_dir.path().join("config"); + let object_root = temp_dir.path().join("objects"); + create_dir(&config_dir)?; + create_dir(&object_root)?; + + let root = toml::Value::String(object_root.display().to_string()); + write_config(&config_dir, "fs", &format!("root = {root}\n"), 1, "2s")?; + + Ok(Self { + _temp_dir: temp_dir, + object_root, + config_dir, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap<String, String> { + HashMap::from([( + CONFIG_DIR_ENV.to_string(), + self.config_dir.display().to_string(), + )]) + } +} + +pub struct OpenDalS3SinkFixture { + floci: FlociContainer, + bucket: Box<Bucket>, + _temp_dir: TempDir, + config_dir: PathBuf, +} + +impl S3SinkOps for OpenDalS3SinkFixture { + fn bucket(&self) -> &Bucket { + &self.bucket + } + + fn endpoint(&self) -> &str { + &self.floci.endpoint + } +} + +#[async_trait] +impl TestFixture for OpenDalS3SinkFixture { + async fn setup() -> Result<Self, TestBinaryError> { + let floci = + FlociContainer::start(None, &fixtures::unique_container_name("floci-opendal")).await?; + let bucket = floci::create_bucket(&floci.endpoint, TEST_BUCKET).await?; + let temp_dir = TempDir::new().map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "OpenDalS3SinkFixture".to_string(), + message: format!("Failed to create temporary directory: {error}"), + })?; + let config_dir = temp_dir.path().join("config"); + create_dir(&config_dir)?; + + let endpoint = toml::Value::String(floci.endpoint.clone()); + let options = format!( + r#"bucket = "{TEST_BUCKET}" +region = "{REGION}" +endpoint = {endpoint} +access_key_id = "{ACCESS_KEY}" +secret_access_key = "{SECRET_KEY}" +"# + ); + write_config(&config_dir, "s3", &options, 1, "2s")?; + + Ok(Self { + floci, + bucket, + _temp_dir: temp_dir, + config_dir, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap<String, String> { + HashMap::from([( + CONFIG_DIR_ENV.to_string(), + self.config_dir.display().to_string(), + )]) + } +} + +pub struct OpenDalControlledS3SinkFixture { + _server: MockServer, + state: Arc<Mutex<S3WriteState>>, + _temp_dir: TempDir, + config_dir: PathBuf, +} + +impl OpenDalControlledS3SinkFixture { + pub fn fail_next_writes(&self, count: usize) { + self.state.lock().expect("S3 state lock").response = S3WriteResponse::Temporary(count); + } + + pub fn deny_writes(&self) { + self.state.lock().expect("S3 state lock").response = S3WriteResponse::Permanent; + } + + pub fn delay_writes(&self, delay: Duration) { + self.state.lock().expect("S3 state lock").response = S3WriteResponse::Delayed(delay); + } + + pub fn object_requests(&self) -> Vec<Request> { + self.state.lock().expect("S3 state lock").requests.clone() + } + + pub fn first_delayed_response_at(&self) -> Option<Instant> { + self.state + .lock() + .expect("S3 state lock") + .first_delayed_response_at + } +} + +#[async_trait] +impl TestFixture for OpenDalControlledS3SinkFixture { + async fn setup() -> Result<Self, TestBinaryError> { + let server = MockServer::start().await; + let state = Arc::new(Mutex::new(S3WriteState::default())); + Mock::given(method("PUT")) + .respond_with(S3WriteResponder(state.clone())) + .mount(&server) + .await; + Mock::given(method("DELETE")) + .respond_with(ResponseTemplate::new(204)) + .mount(&server) + .await; + + let temp_dir = TempDir::new().map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "OpenDalControlledS3SinkFixture".to_string(), + message: format!("Failed to create temporary directory: {error}"), + })?; + let config_dir = temp_dir.path().join("config"); + create_dir(&config_dir)?; + let endpoint = toml::Value::String(server.uri()); + let options = format!( + r#"bucket = "{TEST_BUCKET}" +region = "{REGION}" +endpoint = {endpoint} +access_key_id = "{ACCESS_KEY}" +secret_access_key = "{SECRET_KEY}" +"# + ); + write_config(&config_dir, "s3", &options, 3, "500ms")?; + + Ok(Self { + _server: server, + state, + _temp_dir: temp_dir, + config_dir, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap<String, String> { + HashMap::from([( + CONFIG_DIR_ENV.to_string(), + self.config_dir.display().to_string(), + )]) + } +} + +#[derive(Default)] +struct S3WriteState { + response: S3WriteResponse, + requests: Vec<Request>, + first_delayed_response_at: Option<Instant>, +} + +#[derive(Default)] +enum S3WriteResponse { + #[default] + Success, + Temporary(usize), + Permanent, + Delayed(Duration), +} + +struct S3WriteResponder(Arc<Mutex<S3WriteState>>); + +impl Respond for S3WriteResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let success = ResponseTemplate::new(200) + .insert_header("ETag", "\"d41d8cd98f00b204e9800998ecf8427e\""); + // Startup probes must not consume the object write failure plan. + if !request.url.path().ends_with(".jsonl") { Review Comment: This responder returns success for every PUT that is not a `.jsonl` object and for every DELETE, and no unit test fails the probe either, so all tests stay green with `check_access` deleted. The controlled fixture also runs every case with a 500 ms `write_timeout` (`core/integration/tests/connectors/fixtures/opendal.rs:232`) while the retry and permanent-error tests assert an exact PUT count (`core/integration/tests/connectors/opendal/opendal_sink.rs:138`, `core/integration/tests/connectors/opendal/opendal_sink.rs:180`), so one local response slower than 500 ms on a loaded runner is retried as a temporary timeout and fails the count. No test loads the shipped `core/connectors/sinks/opendal_sink/config.toml`, since `sink.toml` points `config_dir` at a placeholder and each fixture writes its own file. ########## core/connectors/sinks/opendal_sink/src/config.rs: ########## @@ -0,0 +1,617 @@ +// 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. + +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; +use std::{collections::BTreeMap, fmt, path::Path, time::Duration}; + +use iggy_connector_sdk::{ + Error, + retry::{RetryPolicy, parse_duration}, +}; +use opendal::Operator; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tokio::time::timeout; + +use crate::formatter::Formatter; + +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); +const DEFAULT_MAX_FILE_SIZE: &str = "8MiB"; +const DEFAULT_WRITE_TIMEOUT: &str = "60s"; +const MAX_OBJECT_SIZE: u64 = 5000 * 1024 * 1024; + +#[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 max_file_size: Option<String>, + #[serde(default)] + pub write_timeout: Option<String>, + #[serde(default)] + pub if_not_exists: Option<bool>, + #[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("max_file_size", &self.max_file_size) + .field("write_timeout", &self.write_timeout) + .field("if_not_exists", &self.if_not_exists) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub(crate) struct ValidatedConfig { + pub(crate) service: String, + pub(crate) path_prefix: String, + pub(crate) path_template: String, + pub(crate) options: BTreeMap<String, SecretString>, + pub(crate) max_file_size: usize, + pub(crate) output_format: OutputFormat, + pub(crate) include_metadata: bool, + pub(crate) include_headers: bool, + pub(crate) retry_policy: RetryPolicy, + pub(crate) write_timeout: Duration, + pub(crate) if_not_exists: bool, + pub(crate) verbose_logging: bool, +} + +#[derive(Debug)] +pub(crate) enum Validation { + Valid(ValidatedConfig), + Invalid { + config: OpenDalSinkConfig, + error: Error, + }, +} + +impl OpenDalSinkConfig { + pub(crate) fn validate(self) -> Validation { + let checked = (|| { + let mut service = self.service.trim().to_ascii_lowercase(); + if service == "file" { + service.replace_range(.., "fs"); + } + if service.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL service cannot be empty".to_string(), + )); + } + if service == "memory" { + return Err(Error::InvalidConfigValue( + "OpenDAL memory storage does not preserve archived messages".to_string(), + )); + } + if !self.path_template.contains("{stream}") || !self.path_template.contains("{topic}") { + return Err(Error::InvalidConfigValue( + "OpenDAL path_template must contain {stream} and {topic}".to_string(), + )); + } + let max_file_size = self + .max_file_size + .as_deref() + .unwrap_or(DEFAULT_MAX_FILE_SIZE); + let max_file_size = max_file_size + .parse::<byte_unit::Byte>() + .map_err(|error| { + Error::InvalidConfigValue(format!("Invalid max_file_size: {error}")) + })? + .as_u64(); + if max_file_size == 0 || max_file_size > MAX_OBJECT_SIZE { + return Err(Error::InvalidConfigValue( + "OpenDAL max_file_size must be between 1 byte and 5000 MiB".to_string(), + )); + } + let max_file_size = usize::try_from(max_file_size).map_err(|_| { + Error::InvalidConfigValue( + "OpenDAL max_file_size exceeds addressable memory".to_string(), + ) + })?; + let write_timeout = Review Comment: An unparseable `write_timeout` such as `"60"` makes the SDK `parse_duration` fall back to 1 s, not to the 60 s default it is given (`core/connectors/sdk/src/retry.rs:153`), so a missing unit silently cuts every attempt to 1 s plus 1 s per MiB and each batch that misses it is dropped, with only a warn line at startup. This validator already rejects `0s` but lets the typo through, while `s3_sink` rejects an unparseable duration with `InvalidConfigValue` (`core/connectors/sinks/s3_sink/src/lib.rs:276`) and `http_sink` falls back to the declared default (`core/connectors/sinks/http_sink/src/lib.rs:921`). ########## core/connectors/sinks/opendal_sink/src/config.rs: ########## @@ -0,0 +1,617 @@ +// 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. + +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; +use std::{collections::BTreeMap, fmt, path::Path, time::Duration}; + +use iggy_connector_sdk::{ + Error, + retry::{RetryPolicy, parse_duration}, +}; +use opendal::Operator; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tokio::time::timeout; + +use crate::formatter::Formatter; + +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); +const DEFAULT_MAX_FILE_SIZE: &str = "8MiB"; +const DEFAULT_WRITE_TIMEOUT: &str = "60s"; +const MAX_OBJECT_SIZE: u64 = 5000 * 1024 * 1024; + +#[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>, Review Comment: Every backend setting lives in this nested map, and the runtime's env overrides only insert top-level `plugin_config` keys (`core/connectors/runtime/src/configs/connectors/local_provider.rs:405`) from values that are never parsed into an object (`core/configs/src/configs_impl/parsing.rs:57`). `IGGY_CONNECTORS_SINK_OPENDAL_PLUGIN_CONFIG_OPTIONS_BUCKET` becomes an unknown top-level key that serde drops without a word, so the sink keeps writing to the bucket in the file while the operator believes it was overridden. OpenDAL still reads the S3 region and endpoint and the AWS, Azure and GCS credentials from the vendors' own environment variables, but the bucket, the Azure container and the fs `root` can only come from the TOML file, while `s3_sink` takes its bucket from an override (`core/integration/tests/connectors/fixtures/s3/fixture.rs:37`). ########## core/connectors/sinks/opendal_sink/src/lib.rs: ########## @@ -0,0 +1,1066 @@ +// 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::{ + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use async_trait::async_trait; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, retry::retry_async, + sink_connector, +}; +use opendal::{Buffer, Operator}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +use crate::{ + config::{OpenDalSinkConfig, ValidatedConfig, Validation}, + path::{PathContext, object_path}, +}; + +mod config; +mod formatter; +mod path; + +sink_connector!(OpenDalSink); + +const CONNECTOR_NAME: &str = "OpenDAL sink"; +const ABORT_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Debug)] +pub struct OpenDalSink { + id: u32, + state: Option<SinkState>, + messages_processed: AtomicU64, + write_errors: AtomicU64, +} + +#[derive(Debug)] +enum SinkState { + Config(OpenDalSinkConfig), + Running(RunningSink), +} + +#[derive(Debug)] +struct RunningSink { + config: ValidatedConfig, + operator: Operator, + formatter: formatter::Formatter, +} + +impl OpenDalSink { + pub fn new(id: u32, config: OpenDalSinkConfig) -> Self { + Self { + id, + state: Some(SinkState::Config(config)), + messages_processed: AtomicU64::new(0), + write_errors: AtomicU64::new(0), + } + } +} + +impl RunningSink { + async fn write_object(&self, id: u32, path: &str, buffer: Buffer) -> Result<(), Error> { + let retry_context = format!("{CONNECTOR_NAME} connector ID {id} object write"); + retry_async( + self.config.retry_policy, + &retry_context, + opendal::Error::is_temporary, + || self.write_attempt(id, path, buffer.clone()), + ) + .await + .map_err(|failure| { + Error::CannotStoreData(format!( + "Failed to write OpenDAL object '{path}' after {} attempt(s): {}", + failure.attempts, failure.error + )) + }) + } + + async fn write_attempt(&self, id: u32, path: &str, buffer: Buffer) -> opendal::Result<()> { + let duration = self.write_timeout_for(buffer.len()); + let deadline = Instant::now() + duration; + let mut writer = timeout_at( + deadline, + self.operator + .writer_with(path) + .if_not_exists(self.config.if_not_exists), + ) + .await + .map_err(|_| write_timeout_error())??; + let result = timeout_at(deadline, async { + writer.write(buffer).await?; + writer.close().await?; + Ok(()) + }) + .await + .unwrap_or_else(|_| Err(write_timeout_error())); + + if result.is_err() { + match timeout(ABORT_TIMEOUT, writer.abort()).await { + Ok(Ok(())) => {} + Ok(Err(abort_error)) => warn!( + "{CONNECTOR_NAME} connector ID: {id} could not abort object '{path}': {abort_error}" + ), + Err(_) => { + warn!("{CONNECTOR_NAME} connector ID: {id} timed out aborting object '{path}'") + } + } + } + result + } + + /// Adds one second per MiB to the base timeout, scaled to the buffer size. + fn write_timeout_for(&self, buffer_size: usize) -> Duration { + self.config + .write_timeout + .saturating_add(Duration::from_micros( + (buffer_size as u64).saturating_mul(1_000_000) / (1024 * 1024), + )) + } + + async fn check_access(&self, id: u32) -> Result<(), Error> { + let probe_name = format!(".iggy-write-probe-{}", Uuid::new_v4()); + let path_prefix = &self.config.path_prefix; + let path = if path_prefix.is_empty() { + probe_name + } else { + format!("{path_prefix}/{probe_name}") + }; + let write_result = self.write_object(id, &path, Buffer::from("iggy")).await; + let delete_result = timeout(self.config.write_timeout, self.operator.delete(&path)).await; + write_result + .map_err(|error| Error::InitError(format!("OpenDAL write probe failed: {error}")))?; + match delete_result { + Ok(Ok(())) => Ok(()), + Ok(Err(error)) => Err(Error::InitError(format!( + "OpenDAL write probe could not be deleted: {error}" + ))), + Err(_) => Err(Error::InitError( + "OpenDAL write probe deletion timed out".to_string(), + )), + } + } +} + +#[async_trait] +impl Sink for OpenDalSink { + async fn open(&mut self) -> Result<(), Error> { + let config = match self.state.take() { + Some(SinkState::Config(config)) => config, + Some(running @ SinkState::Running(_)) => { Review Comment: `open` never runs twice on one instance. `iggy_sink_open` builds a new sink through `<$type>::new` on every call, refuses an id that is already open and drops the sink when `open` fails (`core/connectors/sdk/src/sink.rs:268`), and `iggy_sink_close` removes it before `close`. The `Running` early return, `Validation::Invalid` handing the raw config back for a second `open`, the `SinkState::Config` round trip, the tests that reopen or consume on an unopened sink and the README paragraph at `core/connectors/sinks/opendal_sink/README.md:194` cover call sequences the runtime cannot produce. A config field plus `Option<RunningSink>` holds every reachable state, and `validate(&self) -> Result<ValidatedConfig, Error>` drops the closure and the six-element tuple. ########## core/connectors/sinks/opendal_sink/src/config.rs: ########## @@ -0,0 +1,617 @@ +// 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. + +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; +use std::{collections::BTreeMap, fmt, path::Path, time::Duration}; + +use iggy_connector_sdk::{ + Error, + retry::{RetryPolicy, parse_duration}, +}; +use opendal::Operator; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use tokio::time::timeout; + +use crate::formatter::Formatter; + +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); +const DEFAULT_MAX_FILE_SIZE: &str = "8MiB"; +const DEFAULT_WRITE_TIMEOUT: &str = "60s"; +const MAX_OBJECT_SIZE: u64 = 5000 * 1024 * 1024; + +#[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 max_file_size: Option<String>, + #[serde(default)] + pub write_timeout: Option<String>, + #[serde(default)] + pub if_not_exists: Option<bool>, + #[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("max_file_size", &self.max_file_size) + .field("write_timeout", &self.write_timeout) + .field("if_not_exists", &self.if_not_exists) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +#[derive(Debug)] +pub(crate) struct ValidatedConfig { + pub(crate) service: String, + pub(crate) path_prefix: String, + pub(crate) path_template: String, + pub(crate) options: BTreeMap<String, SecretString>, + pub(crate) max_file_size: usize, + pub(crate) output_format: OutputFormat, + pub(crate) include_metadata: bool, + pub(crate) include_headers: bool, + pub(crate) retry_policy: RetryPolicy, + pub(crate) write_timeout: Duration, + pub(crate) if_not_exists: bool, + pub(crate) verbose_logging: bool, +} + +#[derive(Debug)] +pub(crate) enum Validation { + Valid(ValidatedConfig), + Invalid { + config: OpenDalSinkConfig, + error: Error, + }, +} + +impl OpenDalSinkConfig { + pub(crate) fn validate(self) -> Validation { + let checked = (|| { + let mut service = self.service.trim().to_ascii_lowercase(); + if service == "file" { + service.replace_range(.., "fs"); + } + if service.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenDAL service cannot be empty".to_string(), + )); + } + if service == "memory" { + return Err(Error::InvalidConfigValue( + "OpenDAL memory storage does not preserve archived messages".to_string(), + )); + } + if !self.path_template.contains("{stream}") || !self.path_template.contains("{topic}") { Review Comment: This only checks that `{stream}` and `{topic}` appear somewhere, and `encode_path_segment` keeps `_` and `-` as they are, so `{stream}_{topic}/{date}` renders stream `a_b` with topic `c` and stream `a` with topic `b_c` to the same `a_b_c` directory. Both topics start at offset 0 on partition 0, so their first objects get the same key and the later write replaces the earlier one. -- 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]
