slbotbm commented on code in PR #3811: URL: https://github.com/apache/iggy/pull/3811#discussion_r3758638922
########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,388 @@ +// 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_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}, + types::AMQPValue, + types::FieldTable, +}; +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: String, Review Comment: (also applies to `RabbitMQSinkConfig::amqp_url`) AMQP URLs normally include the username and password. This field is a `String` on a `Serialize` config type, so the sink's derived `Debug` output and any serialization of this type contain the URL. The runtime also exposes the raw `plugin_config` through its sink-plugin-config endpoint. Store the URL as `secrecy::SecretString`, add `iggy_common::serde_secret::serialize_secret`, and use `ExposeSecret` only at the `Connection::connect` call sites, following the existing Postgres and MongoDB sinks. This prevents plugin-side logging and serialization leaks, but does not redact the runtime's raw `plugin_config`; that endpoint also needs a general redaction mechanism or access restriction. ########## core/connectors/sinks/rabbitmq_sink/README.md: ########## Review Comment: The README documents only the connection and routing fields. It omits `include_metadata`, `verbose_logging`, `max_retries`, `retry_delay_secs`, and `max_retry_delay_secs`, even though all are public plugin configuration. Operators therefore cannot discover how to disable generated headers or control retry behavior. Document each supported field, its type, default, and behavior in the configuration table and TOML example. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,388 @@ +// 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_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}, + types::AMQPValue, + types::FieldTable, +}; +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: String, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + 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")] + amqp_url: String, + #[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>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> String { + "amqp://guest:guest@localhost:5672".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 { + 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), + 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), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + + 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; + let mut published: u64 = 0; + for message in messages { + let body = message.payload.clone().try_into_vec()?; + let mut props = BasicProperties::default(); + if self.include_metadata { + let mut headers = FieldTable::default(); + 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::LongUInt(message.offset.try_into().unwrap_or(u32::MAX)), + ); + props = props.with_headers(headers); + } + + let confirm = channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + .map_err(|e| Error::CannotStoreData(e.to_string()))?; + match confirm.await { + Ok(_) => published += 1, + Err(e) => { + last_error = Some(Error::CannotStoreData(format!("publish rejected: {e}"))); + break; + } + } + } + + if last_error.is_none() { + return Ok(published); + } + + let error = last_error.unwrap(); + attempts += 1; + + if !is_publish_retryable(&error) || attempts >= self.max_retries { + self.publish_errors + .fetch_add(messages.len() as u64 - published, 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() 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; + } + } + + async fn reconnect(&self) -> Result<(), Error> { Review Comment: (also applies to `RabbitMQSink::open`) `ExchangeDeclareOptions::default()` declares a non-durable exchange. RabbitMQ rejects a declaration when an existing exchange with the same name has different durability attributes, closing the channel with `PRECONDITION_FAILED`. Consequently, a normal pre-created durable exchange cannot be used with this connector; the fixture masks this by declaring the same non-durable exchange. Expose the declaration properties in the connector configuration with safe defaults, or declare an operator-managed exchange passively. Cover an existing durable exchange in the integration tests. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## Review Comment: The sink sets `BasicPublishOptions::mandatory = true`, but increments `published` for every `Ok(_)` publisher confirmation. RabbitMQ acknowledges an unroutable mandatory publish with `Confirmation::Ack(Some(returned_message))`; the message has been returned, not routed to a queue. This code therefore returns `Ok(())` and lets the consumed Iggy message advance even though RabbitMQ delivered it nowhere. Match the confirmation explicitly: only `Ack(None)` is success. Treat `Ack(Some(_))` and `Nack(_)` as errors, and add an integration test with a routing key that has no matching binding. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,388 @@ +// 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_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}, + types::AMQPValue, + types::FieldTable, +}; +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: String, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + 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")] + amqp_url: String, + #[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>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> String { + "amqp://guest:guest@localhost:5672".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 { + 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), + 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), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + + 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; + let mut published: u64 = 0; + for message in messages { + let body = message.payload.clone().try_into_vec()?; Review Comment: `consume` only has a borrowed message, but this clones the complete `Payload` before converting it into bytes. For `Payload::Json`, that needlessly deep-clones the `simd_json::OwnedValue` tree before serializing it, adding allocation and CPU cost to every published JSON message. Use `message.payload.try_to_bytes()` to serialize JSON directly from the borrowed payload. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,388 @@ +// 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_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}, + types::AMQPValue, + types::FieldTable, +}; +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: String, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + 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")] + amqp_url: String, + #[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>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> String { + "amqp://guest:guest@localhost:5672".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 { + 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), + 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), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + + 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; + let mut published: u64 = 0; + for message in messages { + let body = message.payload.clone().try_into_vec()?; + let mut props = BasicProperties::default(); + if self.include_metadata { Review Comment: When a later message fails, `published` records how many earlier messages were confirmed, but the next loop iterates over the whole `messages` slice again. A transient error on the final message of a 100-message batch therefore republishes the first 99 confirmed messages. This duplication is introduced by the sink’s own retry loop, independent of any runtime retry behavior. Resume at the first unconfirmed message after reconnecting, and document the remaining at-least-once case where connection loss makes the final publish outcome unknowable. Add a test that forces a failure after at least one confirmation. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## @@ -0,0 +1,388 @@ +// 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_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}, + types::AMQPValue, + types::FieldTable, +}; +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: String, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + 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")] + amqp_url: String, + #[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>, +} + +fn default_exchange_type() -> Option<String> { + Some("topic".into()) +} + +fn default_amqp_url() -> String { + "amqp://guest:guest@localhost:5672".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 { + 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), + 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), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result<u64, Error> { + let mut attempts = 0u32; + + 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; + let mut published: u64 = 0; + for message in messages { + let body = message.payload.clone().try_into_vec()?; + let mut props = BasicProperties::default(); + if self.include_metadata { + let mut headers = FieldTable::default(); + 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::LongUInt(message.offset.try_into().unwrap_or(u32::MAX)), + ); + props = props.with_headers(headers); + } + + let confirm = channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + .map_err(|e| Error::CannotStoreData(e.to_string()))?; Review Comment: An error from `channel.basic_publish(...).await` is converted with `?` and returned immediately. It never sets `last_error`, reconnects, or uses the configured retry delay. Only errors while awaiting a publisher confirmation reach the retry path. Route immediate publish errors through the same retry flow as confirmation errors. Preserve the index of the first unconfirmed message when retrying so this fix does not republish earlier confirmed messages. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## Review Comment: The sink constructs a new `FieldTable` containing only a few generated metadata values and never reads `message.headers`. User headers are therefore lost on every publish. A headers exchange can route on the generated `iggy_*` values when metadata is enabled, but it cannot route on the original user-supplied headers. Encode representable `message.headers` values into AMQP headers, using `ByteArray` for raw binary values rather than a lossy string conversion, then add an integration test that publishes a message with a custom header and routes it through a headers exchange. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## Review Comment: `BasicProperties::default()` leaves the AMQP delivery mode unset, which RabbitMQ treats as non-persistent. There is no configuration to request persistent delivery. Even if the exchange and queue are durable, RabbitMQ can discard a publisher-confirmed message on broker restart, while the sink has already reported it as successfully published. Set persistent delivery mode by default or make it an explicit, documented configuration option. Test restart behavior with a durable exchange and queue. ########## core/connectors/sinks/rabbitmq_sink/src/lib.rs: ########## Review Comment: `ConsumedMessage::offset` is `u64`, but the code converts it to `u32` and silently substitutes `u32::MAX` on overflow. Long-lived topics will consequently publish an incorrect offset header for every message after the first 4,294,967,295 offsets. `lapin` does not expose an `AMQPValue::LongLongUInt`, so encode the full `u64` offset as a decimal `LongString` rather than narrowing it. Cover an offset above `u32::MAX` in a unit test. -- 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]
