numinnex commented on code in PR #3973: URL: https://github.com/apache/iggy/pull/3973#discussion_r3870460476
########## core/connectors/sinks/rabbitmq_sink/README.md: ########## @@ -0,0 +1,38 @@ +# RabbitMQ Sink + +The RabbitMQ sink connector publishes messages from Iggy streams to RabbitMQ exchanges via AMQP 0.9.1. + +## Configuration + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `amqp_url` | string | `amqp://guest:guest@localhost:5672` | RabbitMQ connection URL, including credentials. Treated as a secret: never logged or serialized verbatim. | +| `exchange` | string | `iggy_events` | Exchange name to publish to. | +| `exchange_type` | string | `topic` | Exchange type: `direct`, `topic`, `fanout`, `headers`. | +| `routing_key` | string | `iggy.messages` | Routing key for published messages. | +| `durable_exchange` | bool | `true` | Declare the exchange as durable. Must match the durability of an operator-pre-created exchange, otherwise RabbitMQ closes the channel with `PRECONDITION_FAILED`. | +| `delivery_mode` | string | `persistent` | AMQP delivery mode: `persistent` (2) or `non_persistent` (1). Persistent messages survive broker restarts when the exchange and queue are durable; it forces an fsync on publish, so non-durable topologies may prefer `non_persistent`. | +| `include_metadata` | bool | `true` | Add `iggy_stream`, `iggy_topic`, `iggy_partition_id`, `iggy_offset` message headers. User-supplied headers are always preserved, regardless of this flag. | +| `verbose_logging` | bool | `false` | Log each published batch at `info` level instead of `debug`. | +| `max_retries` | u32 | `3` | Maximum transient publish retries before failing the batch. | +| `retry_delay_secs` | u64 | `1` | Base retry delay in seconds. | +| `max_retry_delay_secs` | u64 | `5` | Upper bound for exponential backoff. | + +User headers on consumed Iggy messages are forwarded as AMQP headers: string values become AMQP `LongString`, raw binary values become `ByteArray`. This allows routing through a `headers` exchange on original user headers. + +Publishes are confirmed via `ConfirmSelect`. With `mandatory = true`, a message with a routing key that matches no binding is returned by RabbitMQ and the batch fails with a permanent error (delivery is at-least-once: if the connection drops mid-batch, the sink resumes from the first unconfirmed message, so a broker-side outcome may be unknowable and could be delivered more than once). Review Comment: The at-least-once claim doesn't hold as shipped. The runtime commits offsets at poll time (`AutoCommitWhen::PollingMessages`) and discards the plugin's FFI return code, so a permanently-failed batch is dropped and still counted as processed. The real guarantee is at-most-once across batches, with at-least-once only within one. `doris_sink/README.md` has the house wording for this. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,621 @@ +// 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 async_trait::async_trait; +use iggy::prelude::HeaderKind; +use iggy_connector_sdk::retry::{exponential_backoff, jitter}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use lapin::{ + BasicProperties, Channel, Connection, ConnectionProperties, ExchangeKind, + options::{ConfirmSelectOptions, ExchangeDeclareOptions}, + publisher_confirm::Confirmation, + types::{AMQPValue, ByteArray, FieldTable, ShortString}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +sink_connector!(RabbitMQSink); + +#[derive(Debug)] +struct RabbitMqState { + connection: Connection, + channel: Channel, +} + +#[derive(Debug)] +pub struct RabbitMQSink { + id: u32, + amqp_url: SecretString, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + durable_exchange: bool, + delivery_mode: u8, + state: Mutex<Option<RabbitMqState>>, + reconnecting: AtomicBool, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + messages_published: AtomicU64, + publish_errors: AtomicU64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RabbitMQSinkConfig { + #[serde( + default = "default_amqp_url", + serialize_with = "iggy_common::serde_secret::serialize_secret" + )] + amqp_url: SecretString, + #[serde(default)] + exchange: Option<String>, + #[serde(default = "default_exchange_type")] + exchange_type: Option<String>, + #[serde(default)] + routing_key: Option<String>, + #[serde(default = "default_true")] + include_metadata: Option<bool>, + #[serde(default)] + verbose_logging: Option<bool>, + #[serde(default = "default_max_retries")] + max_retries: Option<u32>, + #[serde(default = "default_retry_delay_secs")] + retry_delay_secs: Option<u64>, + #[serde(default = "default_max_retry_delay_secs")] + max_retry_delay_secs: Option<u64>, + #[serde(default = "default_true")] + durable_exchange: Option<bool>, + #[serde(default = "default_delivery_mode")] + delivery_mode: Option<String>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> SecretString { + SecretString::from("amqp://guest:guest@localhost:5672") +} + +fn default_delivery_mode() -> Option<String> { + Some("persistent".into()) +} + +fn default_true() -> Option<bool> { + Some(true) +} + +fn default_max_retries() -> Option<u32> { + Some(3) +} +fn default_retry_delay_secs() -> Option<u64> { + Some(1) +} +fn default_max_retry_delay_secs() -> Option<u64> { + Some(5) +} + +impl RabbitMQSink { + pub fn new(id: u32, config: RabbitMQSinkConfig) -> Self { + let delivery_mode = match config.delivery_mode.as_deref() { + Some("non_persistent") => 1, + Some("persistent") => 2, + Some(other) => { + warn!( + "Unknown delivery_mode: {other}, defaulting to persistent for connector ID: {id}" + ); + 2 + } + None => 2, + }; + RabbitMQSink { + id, + amqp_url: config.amqp_url, + exchange: config.exchange.unwrap_or_else(|| "iggy_events".into()), + exchange_type: config.exchange_type.unwrap_or_else(|| "topic".into()), + routing_key: config.routing_key.unwrap_or_else(|| "iggy.messages".into()), + include_metadata: config.include_metadata.unwrap_or(true), + verbose: config.verbose_logging.unwrap_or(false), + durable_exchange: config.durable_exchange.unwrap_or(true), + delivery_mode, + state: Mutex::new(None), + reconnecting: AtomicBool::new(false), + max_retries: config.max_retries.unwrap_or(3), + retry_delay: Duration::from_secs(config.retry_delay_secs.unwrap_or(1)), + max_retry_delay: Duration::from_secs(config.max_retry_delay_secs.unwrap_or(5)), + messages_published: AtomicU64::new(0), + publish_errors: AtomicU64::new(0), + } + } + + fn exchange_kind(&self) -> Result<ExchangeKind, Error> { + match self.exchange_type.as_str() { + "direct" => Ok(ExchangeKind::Direct), + "topic" => Ok(ExchangeKind::Topic), + "fanout" => Ok(ExchangeKind::Fanout), + "headers" => Ok(ExchangeKind::Headers), + other => Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}. Valid: direct, topic, fanout, headers" + ))), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + let mut confirmed: usize = 0; + + loop { + let channel = { + let guard = self.state.lock().await; + guard + .as_ref() + .map(|s| s.channel.clone()) + .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? + }; + + let mut last_error: Option<Error> = None; + for message in &messages[confirmed..] { + let body = message.payload.try_to_bytes()?; + let mut props = BasicProperties::default().with_delivery_mode(self.delivery_mode); + let headers = self.build_headers(topic_metadata, messages_metadata, message); + if !headers.inner().is_empty() { + props = props.with_headers(headers); + } + + let confirm = match channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + { + Ok(confirm) => confirm, + Err(e) => { + last_error = Some(Error::CannotStoreData(e.to_string())); + break; + } + }; + match confirm.await { + Ok(Confirmation::Ack(None)) => confirmed += 1, + Ok(Confirmation::Ack(Some(_))) | Ok(Confirmation::Nack(_)) => { Review Comment: `Nack` probably shouldn't share this arm with `Ack(Some(_))`. `Ack(Some(_))` is the mandatory-return (unroutable) case, whereas a `Nack` means the broker refused responsibility (internal error, disk alarm) — the one confirm outcome RabbitMQ documents as safe to re-publish. As written it becomes `InvalidRecordValue`, which `is_publish_retryable` rejects, so the batch fails permanently. Splitting the arms with `Nack` retryable would also fix the message, which currently misreports the cause. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,621 @@ +// 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 async_trait::async_trait; +use iggy::prelude::HeaderKind; +use iggy_connector_sdk::retry::{exponential_backoff, jitter}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use lapin::{ + BasicProperties, Channel, Connection, ConnectionProperties, ExchangeKind, + options::{ConfirmSelectOptions, ExchangeDeclareOptions}, + publisher_confirm::Confirmation, + types::{AMQPValue, ByteArray, FieldTable, ShortString}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +sink_connector!(RabbitMQSink); + +#[derive(Debug)] +struct RabbitMqState { + connection: Connection, + channel: Channel, +} + +#[derive(Debug)] +pub struct RabbitMQSink { + id: u32, + amqp_url: SecretString, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + durable_exchange: bool, + delivery_mode: u8, + state: Mutex<Option<RabbitMqState>>, + reconnecting: AtomicBool, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + messages_published: AtomicU64, + publish_errors: AtomicU64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RabbitMQSinkConfig { + #[serde( + default = "default_amqp_url", + serialize_with = "iggy_common::serde_secret::serialize_secret" + )] + amqp_url: SecretString, + #[serde(default)] + exchange: Option<String>, + #[serde(default = "default_exchange_type")] + exchange_type: Option<String>, + #[serde(default)] + routing_key: Option<String>, + #[serde(default = "default_true")] + include_metadata: Option<bool>, + #[serde(default)] + verbose_logging: Option<bool>, + #[serde(default = "default_max_retries")] + max_retries: Option<u32>, + #[serde(default = "default_retry_delay_secs")] + retry_delay_secs: Option<u64>, + #[serde(default = "default_max_retry_delay_secs")] + max_retry_delay_secs: Option<u64>, + #[serde(default = "default_true")] + durable_exchange: Option<bool>, + #[serde(default = "default_delivery_mode")] + delivery_mode: Option<String>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> SecretString { + SecretString::from("amqp://guest:guest@localhost:5672") +} + +fn default_delivery_mode() -> Option<String> { + Some("persistent".into()) +} + +fn default_true() -> Option<bool> { + Some(true) +} + +fn default_max_retries() -> Option<u32> { + Some(3) +} +fn default_retry_delay_secs() -> Option<u64> { + Some(1) +} +fn default_max_retry_delay_secs() -> Option<u64> { + Some(5) +} + +impl RabbitMQSink { + pub fn new(id: u32, config: RabbitMQSinkConfig) -> Self { + let delivery_mode = match config.delivery_mode.as_deref() { + Some("non_persistent") => 1, + Some("persistent") => 2, + Some(other) => { + warn!( + "Unknown delivery_mode: {other}, defaulting to persistent for connector ID: {id}" + ); + 2 + } + None => 2, + }; + RabbitMQSink { + id, + amqp_url: config.amqp_url, + exchange: config.exchange.unwrap_or_else(|| "iggy_events".into()), + exchange_type: config.exchange_type.unwrap_or_else(|| "topic".into()), + routing_key: config.routing_key.unwrap_or_else(|| "iggy.messages".into()), + include_metadata: config.include_metadata.unwrap_or(true), + verbose: config.verbose_logging.unwrap_or(false), + durable_exchange: config.durable_exchange.unwrap_or(true), + delivery_mode, + state: Mutex::new(None), + reconnecting: AtomicBool::new(false), + max_retries: config.max_retries.unwrap_or(3), + retry_delay: Duration::from_secs(config.retry_delay_secs.unwrap_or(1)), + max_retry_delay: Duration::from_secs(config.max_retry_delay_secs.unwrap_or(5)), + messages_published: AtomicU64::new(0), + publish_errors: AtomicU64::new(0), + } + } + + fn exchange_kind(&self) -> Result<ExchangeKind, Error> { + match self.exchange_type.as_str() { + "direct" => Ok(ExchangeKind::Direct), + "topic" => Ok(ExchangeKind::Topic), + "fanout" => Ok(ExchangeKind::Fanout), + "headers" => Ok(ExchangeKind::Headers), + other => Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}. Valid: direct, topic, fanout, headers" + ))), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + let mut confirmed: usize = 0; + + loop { + let channel = { + let guard = self.state.lock().await; + guard + .as_ref() + .map(|s| s.channel.clone()) + .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? + }; + + let mut last_error: Option<Error> = None; + for message in &messages[confirmed..] { + let body = message.payload.try_to_bytes()?; + let mut props = BasicProperties::default().with_delivery_mode(self.delivery_mode); + let headers = self.build_headers(topic_metadata, messages_metadata, message); + if !headers.inner().is_empty() { + props = props.with_headers(headers); + } + + let confirm = match channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + { + Ok(confirm) => confirm, + Err(e) => { + last_error = Some(Error::CannotStoreData(e.to_string())); + break; + } + }; + match confirm.await { + Ok(Confirmation::Ack(None)) => confirmed += 1, + Ok(Confirmation::Ack(Some(_))) | Ok(Confirmation::Nack(_)) => { + last_error = Some(Error::InvalidRecordValue( + "message returned as unroutable by RabbitMQ".into(), + )); + break; + } + Ok(Confirmation::NotRequested) => { + last_error = Some(Error::CannotStoreData( + "publisher confirms not enabled".into(), + )); + break; + } + Err(e) => { + last_error = Some(Error::CannotStoreData(format!("publish rejected: {e}"))); + break; + } + } + } + + if last_error.is_none() { + return Ok(confirmed as u64); + } + + let error = last_error.unwrap(); + attempts += 1; + + if !is_publish_retryable(&error) || attempts >= self.max_retries { + self.publish_errors + .fetch_add((messages.len() - confirmed) as u64, Ordering::Relaxed); + return Err(Error::CannotStoreData(format!( + "batch publish failed after {attempts} attempts: {error}" + ))); + } + + match self.reconnect().await { + Ok(_) => {} + Err(reconnect_error) => { + self.publish_errors + .fetch_add((messages.len() - confirmed) as u64, Ordering::Relaxed); + return Err(Error::Connection(format!( + "failed to reconnect: {reconnect_error}" + ))); + } + } + + let delay = jitter(exponential_backoff( + self.retry_delay, + attempts.saturating_sub(1), + self.max_retry_delay, + )); + warn!( + "Transient RabbitMQ publish error for connector ID: {} (attempt {attempts}/{}): {error}. Retrying in {:?}.", + self.id, self.max_retries, delay + ); + tokio::time::sleep(delay).await; + } + } + + fn build_headers( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + message: &ConsumedMessage, + ) -> FieldTable { + let mut headers = FieldTable::default(); + if let Some(user_headers) = &message.headers + && !user_headers.is_empty() + { + for (key, value) in user_headers { + let name = ShortString::from(key.to_string_value()); + let amqp_value = match value.kind() { + HeaderKind::String => AMQPValue::LongString(value.to_string_value().into()), + _ => AMQPValue::ByteArray(ByteArray::from(value.as_bytes())), + }; + headers.insert(name, amqp_value); + } + } + if self.include_metadata { + headers.insert( + "iggy_stream".into(), + AMQPValue::LongString(topic_metadata.stream.clone().into()), + ); + headers.insert( + "iggy_topic".into(), + AMQPValue::LongString(topic_metadata.topic.clone().into()), + ); + headers.insert( + "iggy_partition_id".into(), + AMQPValue::LongUInt(messages_metadata.partition_id), + ); + headers.insert( + "iggy_offset".into(), + AMQPValue::LongLongInt(message.offset as i64), + ); + } + headers + } + + async fn reconnect(&self) -> Result<(), Error> { + if self + .reconnecting + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + tokio::time::sleep(self.retry_delay).await; + return Ok(()); + } + + warn!("Reconnecting RabbitMQ sink ID: {}", self.id); + let result = async { + let conn = Connection::connect( + self.amqp_url.expose_secret(), + ConnectionProperties::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let channel = conn + .create_channel() + .await + .map_err(|e| Error::Connection(e.to_string()))?; + channel + .confirm_select(ConfirmSelectOptions::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let exchange_kind = self.exchange_kind()?; + channel + .exchange_declare( + &self.exchange, + exchange_kind, + ExchangeDeclareOptions { + durable: self.durable_exchange, + ..Default::default() + }, + FieldTable::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + *self.state.lock().await = Some(RabbitMqState { + connection: conn, + channel, + }); + Ok::<(), Error>(()) + } + .await; + self.reconnecting.store(false, Ordering::Release); + result + } +} + +#[async_trait] +impl Sink for RabbitMQSink { + async fn open(&mut self) -> Result<(), Error> { + let exchange_kind = self.exchange_kind()?; + let conn = Connection::connect( + self.amqp_url.expose_secret(), + ConnectionProperties::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let channel = conn + .create_channel() + .await + .map_err(|e| Error::Connection(e.to_string()))?; + channel + .confirm_select(lapin::options::ConfirmSelectOptions::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + + channel + .exchange_declare( + &self.exchange, + exchange_kind, + ExchangeDeclareOptions { + durable: self.durable_exchange, + ..Default::default() + }, + FieldTable::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + *self.state.get_mut() = Some(RabbitMqState { + connection: conn, + channel, + }); + info!( + "Opened RabbitMQ sink ID: {}, connected to exchange: {}", + self.id, self.exchange + ); + + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let published = self + .publish_batch_with_retry(topic_metadata, &messages_metadata, &messages) + .await?; + self.messages_published + .fetch_add(published, Ordering::Relaxed); + if self.verbose { + info!( + "Published {published} messages to exchange: {}", + self.exchange + ); + } else { + debug!( + "Published {published} messages to exchange: {}", + self.exchange + ); + } + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + let published = self.messages_published.load(Ordering::Relaxed); + let errors = self.publish_errors.load(Ordering::Relaxed); + info!( + "RabbitMQ sink ID: {} processed {} messages with {} errors", + self.id, published, errors + ); + + if let Some(state) = self.state.get_mut().take() { + state + .channel + .close(200, "OK") + .await + .map_err(|e| Error::Connection(e.to_string()))?; + state + .connection + .close(200, "OK") + .await + .map_err(|e| Error::Connection(e.to_string()))?; + } + Ok(()) + } +} + +fn is_publish_retryable(error: &Error) -> bool { Review Comment: These substrings don't match what lapin actually emits. A dropped channel renders as `invalid channel state: Closed` (not `channel closed`), heartbeat loss as `no heartbeat received from server for too long`, AMQP uses `RESOURCE_LOCKED` with an underscore, and `io::ErrorKind::TimedOut` prints `timed out`, not `timeout`. So a channel-level exception is classified permanent and returns at L239 before `reconnect()` is ever called. Since `state` is only written on success (L350), the dead `Channel` is then re-read by every subsequent batch and the sink never self-heals. Suggest matching on `lapin::Error` variants instead, and clearing `state` on any publish failure — the classifier fix alone still wedges on an error the list doesn't cover. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,621 @@ +// 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 async_trait::async_trait; +use iggy::prelude::HeaderKind; +use iggy_connector_sdk::retry::{exponential_backoff, jitter}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use lapin::{ + BasicProperties, Channel, Connection, ConnectionProperties, ExchangeKind, + options::{ConfirmSelectOptions, ExchangeDeclareOptions}, + publisher_confirm::Confirmation, + types::{AMQPValue, ByteArray, FieldTable, ShortString}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +sink_connector!(RabbitMQSink); + +#[derive(Debug)] +struct RabbitMqState { + connection: Connection, + channel: Channel, +} + +#[derive(Debug)] +pub struct RabbitMQSink { + id: u32, + amqp_url: SecretString, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + durable_exchange: bool, + delivery_mode: u8, + state: Mutex<Option<RabbitMqState>>, + reconnecting: AtomicBool, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + messages_published: AtomicU64, + publish_errors: AtomicU64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RabbitMQSinkConfig { + #[serde( + default = "default_amqp_url", + serialize_with = "iggy_common::serde_secret::serialize_secret" + )] + amqp_url: SecretString, + #[serde(default)] + exchange: Option<String>, + #[serde(default = "default_exchange_type")] + exchange_type: Option<String>, + #[serde(default)] + routing_key: Option<String>, + #[serde(default = "default_true")] + include_metadata: Option<bool>, + #[serde(default)] + verbose_logging: Option<bool>, + #[serde(default = "default_max_retries")] + max_retries: Option<u32>, + #[serde(default = "default_retry_delay_secs")] + retry_delay_secs: Option<u64>, + #[serde(default = "default_max_retry_delay_secs")] + max_retry_delay_secs: Option<u64>, + #[serde(default = "default_true")] + durable_exchange: Option<bool>, + #[serde(default = "default_delivery_mode")] + delivery_mode: Option<String>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> SecretString { + SecretString::from("amqp://guest:guest@localhost:5672") +} + +fn default_delivery_mode() -> Option<String> { + Some("persistent".into()) +} + +fn default_true() -> Option<bool> { + Some(true) +} + +fn default_max_retries() -> Option<u32> { + Some(3) +} +fn default_retry_delay_secs() -> Option<u64> { + Some(1) +} +fn default_max_retry_delay_secs() -> Option<u64> { + Some(5) +} + +impl RabbitMQSink { + pub fn new(id: u32, config: RabbitMQSinkConfig) -> Self { + let delivery_mode = match config.delivery_mode.as_deref() { + Some("non_persistent") => 1, + Some("persistent") => 2, + Some(other) => { + warn!( + "Unknown delivery_mode: {other}, defaulting to persistent for connector ID: {id}" + ); + 2 + } + None => 2, + }; + RabbitMQSink { + id, + amqp_url: config.amqp_url, + exchange: config.exchange.unwrap_or_else(|| "iggy_events".into()), + exchange_type: config.exchange_type.unwrap_or_else(|| "topic".into()), + routing_key: config.routing_key.unwrap_or_else(|| "iggy.messages".into()), + include_metadata: config.include_metadata.unwrap_or(true), + verbose: config.verbose_logging.unwrap_or(false), + durable_exchange: config.durable_exchange.unwrap_or(true), + delivery_mode, + state: Mutex::new(None), + reconnecting: AtomicBool::new(false), + max_retries: config.max_retries.unwrap_or(3), + retry_delay: Duration::from_secs(config.retry_delay_secs.unwrap_or(1)), + max_retry_delay: Duration::from_secs(config.max_retry_delay_secs.unwrap_or(5)), + messages_published: AtomicU64::new(0), + publish_errors: AtomicU64::new(0), + } + } + + fn exchange_kind(&self) -> Result<ExchangeKind, Error> { + match self.exchange_type.as_str() { + "direct" => Ok(ExchangeKind::Direct), + "topic" => Ok(ExchangeKind::Topic), + "fanout" => Ok(ExchangeKind::Fanout), + "headers" => Ok(ExchangeKind::Headers), + other => Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}. Valid: direct, topic, fanout, headers" + ))), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + let mut confirmed: usize = 0; + + loop { + let channel = { + let guard = self.state.lock().await; + guard + .as_ref() + .map(|s| s.channel.clone()) + .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? + }; + + let mut last_error: Option<Error> = None; + for message in &messages[confirmed..] { + let body = message.payload.try_to_bytes()?; + let mut props = BasicProperties::default().with_delivery_mode(self.delivery_mode); + let headers = self.build_headers(topic_metadata, messages_metadata, message); + if !headers.inner().is_empty() { + props = props.with_headers(headers); + } + + let confirm = match channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + { + Ok(confirm) => confirm, + Err(e) => { + last_error = Some(Error::CannotStoreData(e.to_string())); + break; + } + }; + match confirm.await { Review Comment: Awaiting the confirm inside the per-message loop serializes a broker round-trip per message, so throughput is capped at 1/RTT regardless of payload or batch size (~2k msg/s on a LAN, well under that cross-AZ). With `batch_length = 100` here and 1000 as the runtime default, that's 100–1000 sequential RTTs per batch, and it also stops lapin from coalescing frames. Worth flagging that the obvious fix isn't safe as-is: lapin pairs `Basic.Return` to a confirm FIFO with no delivery tag, so with several publishes in flight an unroutable message can be attributed to the wrong one. Pipelining would need `mandatory` off on that path, or a per-message `message_id` to correlate on. -- 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]
