github-actions[bot] commented on code in PR #4538: URL: https://github.com/apache/iggy/pull/4538#discussion_r4267500073
########## core/connectors/sinks/redis_sink/README.md: ########## @@ -0,0 +1,54 @@ +# Redis Sink + +Writes messages from Iggy streams into [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/) using pipelined `XADD`. + +## Configuration + +```toml +[plugin_config] +connection_url = "redis://localhost:6379" +mode = "stream" +stream_key = "iggy:events" +batch_size = 100 +payload_format = "json" +include_metadata = true +deterministic_ids = true +max_len = 100000 +max_retries = 3 +retry_delay = "1s" +response_timeout = "5s" +``` + +| Field | Default | Description | +| ------------------- | -------- | ------------------------------------------------------------------------------------------------------------------ | +| `connection_url` | required | Redis connection URL. May carry credentials, never logged. `url` is accepted as a legacy alias | +| `mode` | `stream` | Connector mode. Redis Streams is the only v1 mode | +| `stream_key` | required | Redis Stream key prefix, see [Idempotency](#idempotency) | +| `batch_size` | `100` | Entries per pipeline round trip | +| `payload_format` | `binary` | `binary` stores bytes as-is. `text` requires UTF-8. `json` requires a valid JSON document | +| `include_metadata` | `true` | Adds `message_id`, `offset`, `partition_id`, `stream`, `topic`, `timestamp`, `origin_timestamp`, `checksum` fields | +| `deterministic_ids` | `true` | Derives the entry ID from the message offset | +| `max_len` | none | Approximate `MAXLEN ~` trim applied on every `XADD` | +| `max_retries` | `3` | Total attempts for connection failures, including the first attempt and capped at 3 | +| `retry_delay` | `1s` | Base backoff delay | +| `response_timeout` | `5s` | Time to wait for Redis to answer one pipeline before the attempt counts as a transient failure | Review Comment: nit: The table and the sample omit `verbose_logging`, which the connector's `config.toml` sets and other sink READMEs document. Add the field with its `false` default. ########## core/connectors/sinks/redis_sink/src/lib.rs: ########## @@ -0,0 +1,837 @@ +// 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; +use std::str::FromStr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use humantime::Duration as HumanDuration; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector, +}; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; +use redis::{Client, Pipeline, RedisError, RedisResult}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize, de::IgnoredAny}; +use thiserror::Error as ThisError; +use tracing::{debug, error, info, warn}; + +sink_connector!(RedisSink); + +const CONNECTOR_NAME: &str = "Redis sink"; +const DEFAULT_BATCH_SIZE: u32 = 100; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: Duration = Duration::from_secs(1); +const DEFAULT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(30); +const DUPLICATE_ENTRY_DETAIL: &str = "equal or smaller than the target stream top item"; +const HEADER_ENCODING_BASE64: &str = "base64"; + +// `Deserialize` only: nothing re-serializes a plugin config, and leaving +// `Serialize` off keeps the URL, which may carry a password, unserializable. +#[derive(Debug, Clone, Deserialize)] +pub struct RedisSinkConfig { + #[serde(alias = "url")] + pub connection_url: SecretString, + pub mode: Option<RedisMode>, + pub stream_key: String, + pub batch_size: Option<u32>, + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub deterministic_ids: Option<bool>, + pub max_len: Option<u64>, + pub verbose_logging: Option<bool>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub response_timeout: Option<String>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum RedisMode { + #[default] + Stream, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PayloadFormat { + #[default] + Binary, + Text, + Json, +} + +impl From<Option<&str>> for PayloadFormat { + fn from(value: Option<&str>) -> Self { + match value.map(str::to_ascii_lowercase).as_deref() { + None | Some("binary") => PayloadFormat::Binary, + Some("text") => PayloadFormat::Text, + Some("json") => PayloadFormat::Json, + Some(unknown) => { + warn!("Unknown {CONNECTOR_NAME} payload_format '{unknown}', defaulting to binary"); + PayloadFormat::Binary + } + } + } +} + +#[derive(Debug, ThisError)] +enum EntryError { + #[error("failed to serialize JSON payload: {0}")] + JsonPayload(Error), + #[error("payload is not valid UTF-8: {0}")] + Utf8(#[from] std::str::Utf8Error), + #[error("payload is not valid JSON: {0}")] + InvalidJson(serde_json::Error), + #[error("failed to encode headers: {0}")] + Headers(serde_json::Error), +} + +#[derive(Serialize)] +struct EncodedHeader { + data: String, + iggy_header_encoding: &'static str, +} + +struct RedisEntry<'a> { + message: &'a ConsumedMessage, + payload: &'a [u8], + headers: Option<&'a [u8]>, +} + +#[derive(Debug)] +pub struct RedisSink { + pub id: u32, + connection: Option<ConnectionManager>, + config: RedisSinkConfig, + verbose: bool, + batch_size: usize, + payload_format: PayloadFormat, + include_metadata: bool, + deterministic_ids: bool, + retry_policy: RetryPolicy, + response_timeout: Duration, + messages_processed: AtomicU64, + duplicates_skipped: AtomicU64, + errors: AtomicU64, +} + +impl RedisSink { + pub fn new(id: u32, config: RedisSinkConfig) -> Self { + let retry_delay = parse_duration( + config.retry_delay.as_deref(), + "retry_delay", + DEFAULT_RETRY_DELAY, + id, + ); + let response_timeout = parse_duration( + config.response_timeout.as_deref(), + "response_timeout", + DEFAULT_RESPONSE_TIMEOUT, + id, + ); + RedisSink { + id, + connection: None, + verbose: config.verbose_logging.unwrap_or(false), + batch_size: config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize, + payload_format: PayloadFormat::from(config.payload_format.as_deref()), + include_metadata: config.include_metadata.unwrap_or(true), + deterministic_ids: config.deterministic_ids.unwrap_or(true), + retry_policy: RetryPolicy { + max_attempts: config + .max_retries + .unwrap_or(DEFAULT_MAX_RETRIES) + .clamp(1, DEFAULT_MAX_RETRIES), + base_delay: retry_delay, + max_delay: MAX_RETRY_DELAY.max(retry_delay), + }, + response_timeout, + config, + messages_processed: AtomicU64::new(0), + duplicates_skipped: AtomicU64::new(0), + errors: AtomicU64::new(0), + } + } +} + +#[async_trait] +impl Sink for RedisSink { + async fn open(&mut self) -> Result<(), Error> { + let client = Client::open(self.config.connection_url.expose_secret()) + .map_err(|error| Error::InitError(format!("invalid Redis URL: {error}")))?; + let manager_config = + ConnectionManagerConfig::new().set_response_timeout(Some(self.response_timeout)); + let mut connection = ConnectionManager::new_with_config(client, manager_config) + .await + .map_err(|error| Error::InitError(format!("Redis connection failed: {error}")))?; + redis::cmd("PING") + .query_async::<String>(&mut connection) + .await + .map_err(|error| Error::InitError(format!("Redis PING failed: {error}")))?; + self.connection = Some(connection); + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, stream_key: {}", + self.id, self.config.stream_key + ); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let Some(connection) = self.connection.as_ref() else { + return Err(Error::InitError("Redis connection not initialized".into())); + }; + + if self.verbose { + info!( + "{CONNECTOR_NAME} ID: {} consuming {} messages from stream: {}, topic: {}, partition_id: {}, current_offset: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset + ); + } else { + debug!( + "{CONNECTOR_NAME} ID: {} consuming {} messages", + self.id, + messages.len() + ); + } + + let stream_key = self.target_key(topic_metadata, messages_metadata.partition_id); + for batch in messages.chunks(self.batch_size) { + if let Err(error) = self + .write_batch( + connection, + &stream_key, + topic_metadata, + &messages_metadata, + batch, + ) + .await + { + error!( + "{CONNECTOR_NAME} ID: {} failed to write batch to {stream_key}: {error}", + self.id + ); + return Err(error); Review Comment: critical: `consume()` stops at the first failed chunk, so the later chunks of that call are never attempted. The runtime commits offsets at poll, so those messages are lost. Write every chunk, keep the first error, and return it after the loop, as `postgres_sink` and `quickwit_sink` do. ########## core/connectors/sinks/redis_sink/src/lib.rs: ########## @@ -0,0 +1,837 @@ +// 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; +use std::str::FromStr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use humantime::Duration as HumanDuration; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector, +}; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; +use redis::{Client, Pipeline, RedisError, RedisResult}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize, de::IgnoredAny}; +use thiserror::Error as ThisError; +use tracing::{debug, error, info, warn}; + +sink_connector!(RedisSink); + +const CONNECTOR_NAME: &str = "Redis sink"; +const DEFAULT_BATCH_SIZE: u32 = 100; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: Duration = Duration::from_secs(1); +const DEFAULT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(30); +const DUPLICATE_ENTRY_DETAIL: &str = "equal or smaller than the target stream top item"; +const HEADER_ENCODING_BASE64: &str = "base64"; + +// `Deserialize` only: nothing re-serializes a plugin config, and leaving +// `Serialize` off keeps the URL, which may carry a password, unserializable. +#[derive(Debug, Clone, Deserialize)] +pub struct RedisSinkConfig { + #[serde(alias = "url")] + pub connection_url: SecretString, + pub mode: Option<RedisMode>, + pub stream_key: String, + pub batch_size: Option<u32>, + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub deterministic_ids: Option<bool>, + pub max_len: Option<u64>, + pub verbose_logging: Option<bool>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub response_timeout: Option<String>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum RedisMode { + #[default] + Stream, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PayloadFormat { + #[default] + Binary, + Text, + Json, +} + +impl From<Option<&str>> for PayloadFormat { + fn from(value: Option<&str>) -> Self { + match value.map(str::to_ascii_lowercase).as_deref() { + None | Some("binary") => PayloadFormat::Binary, + Some("text") => PayloadFormat::Text, + Some("json") => PayloadFormat::Json, + Some(unknown) => { + warn!("Unknown {CONNECTOR_NAME} payload_format '{unknown}', defaulting to binary"); + PayloadFormat::Binary + } + } + } +} + +#[derive(Debug, ThisError)] +enum EntryError { + #[error("failed to serialize JSON payload: {0}")] + JsonPayload(Error), + #[error("payload is not valid UTF-8: {0}")] + Utf8(#[from] std::str::Utf8Error), + #[error("payload is not valid JSON: {0}")] + InvalidJson(serde_json::Error), + #[error("failed to encode headers: {0}")] + Headers(serde_json::Error), +} + +#[derive(Serialize)] +struct EncodedHeader { + data: String, + iggy_header_encoding: &'static str, +} + +struct RedisEntry<'a> { + message: &'a ConsumedMessage, + payload: &'a [u8], + headers: Option<&'a [u8]>, +} + +#[derive(Debug)] +pub struct RedisSink { + pub id: u32, + connection: Option<ConnectionManager>, + config: RedisSinkConfig, + verbose: bool, + batch_size: usize, + payload_format: PayloadFormat, + include_metadata: bool, + deterministic_ids: bool, + retry_policy: RetryPolicy, + response_timeout: Duration, + messages_processed: AtomicU64, + duplicates_skipped: AtomicU64, + errors: AtomicU64, +} + +impl RedisSink { + pub fn new(id: u32, config: RedisSinkConfig) -> Self { + let retry_delay = parse_duration( + config.retry_delay.as_deref(), + "retry_delay", + DEFAULT_RETRY_DELAY, + id, + ); + let response_timeout = parse_duration( + config.response_timeout.as_deref(), + "response_timeout", + DEFAULT_RESPONSE_TIMEOUT, + id, + ); + RedisSink { + id, + connection: None, + verbose: config.verbose_logging.unwrap_or(false), + batch_size: config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize, + payload_format: PayloadFormat::from(config.payload_format.as_deref()), + include_metadata: config.include_metadata.unwrap_or(true), + deterministic_ids: config.deterministic_ids.unwrap_or(true), + retry_policy: RetryPolicy { + max_attempts: config + .max_retries + .unwrap_or(DEFAULT_MAX_RETRIES) + .clamp(1, DEFAULT_MAX_RETRIES), + base_delay: retry_delay, + max_delay: MAX_RETRY_DELAY.max(retry_delay), + }, + response_timeout, + config, + messages_processed: AtomicU64::new(0), + duplicates_skipped: AtomicU64::new(0), + errors: AtomicU64::new(0), + } + } +} + +#[async_trait] +impl Sink for RedisSink { + async fn open(&mut self) -> Result<(), Error> { + let client = Client::open(self.config.connection_url.expose_secret()) + .map_err(|error| Error::InitError(format!("invalid Redis URL: {error}")))?; + let manager_config = + ConnectionManagerConfig::new().set_response_timeout(Some(self.response_timeout)); + let mut connection = ConnectionManager::new_with_config(client, manager_config) + .await + .map_err(|error| Error::InitError(format!("Redis connection failed: {error}")))?; + redis::cmd("PING") + .query_async::<String>(&mut connection) + .await + .map_err(|error| Error::InitError(format!("Redis PING failed: {error}")))?; + self.connection = Some(connection); + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, stream_key: {}", + self.id, self.config.stream_key + ); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let Some(connection) = self.connection.as_ref() else { + return Err(Error::InitError("Redis connection not initialized".into())); + }; + + if self.verbose { + info!( + "{CONNECTOR_NAME} ID: {} consuming {} messages from stream: {}, topic: {}, partition_id: {}, current_offset: {}", + self.id, + messages.len(), + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset + ); + } else { + debug!( + "{CONNECTOR_NAME} ID: {} consuming {} messages", + self.id, + messages.len() + ); + } + + let stream_key = self.target_key(topic_metadata, messages_metadata.partition_id); + for batch in messages.chunks(self.batch_size) { + if let Err(error) = self + .write_batch( + connection, + &stream_key, + topic_metadata, + &messages_metadata, + batch, + ) + .await + { + error!( + "{CONNECTOR_NAME} ID: {} failed to write batch to {stream_key}: {error}", + self.id + ); + return Err(error); + } + } + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + self.connection.take(); + info!( + "Closed {CONNECTOR_NAME} connector ID: {}, processed: {}, duplicates skipped: {}, errors: {}", + self.id, + self.messages_processed.load(Ordering::Relaxed), + self.duplicates_skipped.load(Ordering::Relaxed), + self.errors.load(Ordering::Relaxed) + ); + Ok(()) + } +} + +impl RedisSink { + // Explicit entry IDs must grow monotonically per key, and offsets only do + // within one partition, so each (stream, topic, partition) gets its own key. + fn target_key(&self, topic_metadata: &TopicMetadata, partition_id: u32) -> String { + if self.deterministic_ids { + format!( + "{}:{}:{}:{partition_id}", + self.config.stream_key, topic_metadata.stream, topic_metadata.topic + ) + } else { + self.config.stream_key.clone() + } + } + + async fn write_batch( + &self, + connection: &ConnectionManager, + stream_key: &str, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + batch: &[ConsumedMessage], + ) -> Result<(), Error> { + let mut pipeline = redis::pipe(); + pipeline.ignore_errors(); + let mut queued = 0u64; + let mut invalid = 0u64; + for message in batch { + let (payload, headers) = match self + .payload_bytes(message) + .and_then(|payload| Ok((payload, self.headers_bytes(message)?))) + { + Ok(encoded) => encoded, + Err(reason) => { + error!( + "{CONNECTOR_NAME} ID: {} skipping message offset: {}: {reason}", + self.id, message.offset + ); + invalid += 1; + continue; + } + }; + self.queue_entry( + &mut pipeline, + stream_key, + topic_metadata, + messages_metadata, + RedisEntry { + message, + payload: &payload, + headers: headers.as_deref(), + }, + ); + queued += 1; + } + if invalid > 0 { + self.errors.fetch_add(invalid, Ordering::Relaxed); + } + if queued == 0 { Review Comment: warning: `write_batch()` returns `Ok` after dropping messages that fail the payload or header check, so the runtime counts them as processed and records no error. Return the last entry error after the pipeline, as `quickwit_sink` does, so the drop reaches the runtime error counter. ########## core/connectors/sinks/redis_sink/src/lib.rs: ########## @@ -0,0 +1,837 @@ +// 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; +use std::str::FromStr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use humantime::Duration as HumanDuration; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector, +}; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; +use redis::{Client, Pipeline, RedisError, RedisResult}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize, de::IgnoredAny}; +use thiserror::Error as ThisError; +use tracing::{debug, error, info, warn}; + +sink_connector!(RedisSink); + +const CONNECTOR_NAME: &str = "Redis sink"; +const DEFAULT_BATCH_SIZE: u32 = 100; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: Duration = Duration::from_secs(1); +const DEFAULT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(30); +const DUPLICATE_ENTRY_DETAIL: &str = "equal or smaller than the target stream top item"; +const HEADER_ENCODING_BASE64: &str = "base64"; + +// `Deserialize` only: nothing re-serializes a plugin config, and leaving +// `Serialize` off keeps the URL, which may carry a password, unserializable. +#[derive(Debug, Clone, Deserialize)] +pub struct RedisSinkConfig { + #[serde(alias = "url")] + pub connection_url: SecretString, + pub mode: Option<RedisMode>, + pub stream_key: String, + pub batch_size: Option<u32>, + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub deterministic_ids: Option<bool>, + pub max_len: Option<u64>, + pub verbose_logging: Option<bool>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub response_timeout: Option<String>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum RedisMode { + #[default] + Stream, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PayloadFormat { + #[default] + Binary, + Text, + Json, +} + +impl From<Option<&str>> for PayloadFormat { Review Comment: nit: `PayloadFormat` is built through `From<Option<&str>>`, while the other sinks name this conversion `from_config`. Rename it to `PayloadFormat::from_config` to match them. -- 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]
