feat(connectors): add RabbitMQ sink - #3811
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
❌ Your patch check has failed because the patch coverage (43.31%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## master #3811 +/- ##
============================================
- Coverage 75.62% 75.54% -0.09%
Complexity 1046 1046
============================================
Files 1337 1338 +1
Lines 169125 169312 +187
Branches 141481 141745 +264
============================================
- Hits 127907 127906 -1
- Misses 37435 37527 +92
- Partials 3783 3879 +96
🚀 New features to boost your workflow:
|
|
/request-review @hubcio |
| #[derive(Debug)] | ||
| pub struct RabbitMQSink { | ||
| id: u32, | ||
| amqp_url: String, |
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| async fn reconnect(&self) -> Result<(), Error> { |
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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()))?; |
There was a problem hiding this 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.
| 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 { |
There was a problem hiding this 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.
| let mut last_error: Option<Error> = None; | ||
| let mut published: u64 = 0; | ||
| for message in messages { | ||
| let body = message.payload.clone().try_into_vec()?; |
There was a problem hiding this 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.
|
/author |
github-actions
Bot
added
S-waiting-on-author
2 participants
Which issue does this PR address?
Relates to #3747
Sumary
This change adds the RabbitMQ sink connector via the lapin client. As requested, doing only sink connector in this PR. Source will be a separate one.
Adds Configurable exchange and various types (topic, direct, fanout).
Adds Configurable Retries with exponential backoff
Added Integration tests against RabbitMQ container covering topic, fanout and direct exchange behavior.
Local Execution
Completed below based on https://github.com/apache/iggy/blob/master/CONTRIBUTING.md