feat(connectors): add RabbitMQ sink - #3973
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
| } | ||
| } | ||
|
|
||
| fn is_publish_retryable(error: &Error) -> bool { |
There was a problem hiding this 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.
| }; | ||
| match confirm.await { | ||
| Ok(Confirmation::Ack(None)) => confirmed += 1, | ||
| Ok(Confirmation::Ack(Some(_))) | Ok(Confirmation::Nack(_)) => { |
There was a problem hiding this 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.
| break; | ||
| } | ||
| }; | ||
| match confirm.await { |
There was a problem hiding this 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.
|
|
||
| warn!("Reconnecting RabbitMQ sink ID: {}", self.id); | ||
| let result = async { | ||
| let conn = Connection::connect( |
There was a problem hiding this comment.
No timeout on Connection::connect here, nor on basic_publish / confirm.await in the publish loop. RabbitMQ blocks publishers on a disk or memory alarm while still answering heartbeats, so the confirm can hang indefinitely, and a blackholed SYN here waits out the kernel TCP timeout while holding the reconnecting CAS.
Since the FFI entry point is block_on, that parks a connectors-runtime worker thread. surrealdb_sink and clickhouse_sink both expose a timeout config for this.
|
|
||
| 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). |
There was a problem hiding this 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.
2 participants
Replaces #3811 (could not be reopened after rebasing onto master). Rebased to current master and addressed all review comments.
Which issue does this PR address?
Relates to #3747
Summary
Adds the RabbitMQ sink connector via the lapin client. Source will be a separate PR.
Review feedback addressed
amqp_urlstored assecrecy::SecretString, redacted viaiggy_common::serde_secret(never logged or serialized verbatim)durable_exchangeconfig (defaulttrue)Ack(None)counts as success;Ack(Some(_))/Nack(_)(unroutable mandatory publish) fail the batch permanentlydelivery_modeconfig (defaultpersistent)LongStringfor strings,ByteArrayfor binary) so headers exchanges can route on themiggy_offsetencoded as fulli64instead of narrowing tou32basic_publisherrors routed through the same retry flow; the retry loop resumes at the first unconfirmed message instead of republishing confirmed onesPayload::try_to_bytes()(no deep clone)AI usage
Which tools? opencode
Scope of usage? investigation, code suggestions, implementation of review feedback
How did you verify the generated code works correctly? Read through, compiled, ran unit tests, ran clippy/fmt/sort
Can you explain every line of the code if asked? Yes