hubcio commented on code in PR #4735: URL: https://github.com/apache/iggy/pull/4735#discussion_r4468917199
########## core/connectors/sources/rabbitmq_source/src/message.rs: ########## @@ -0,0 +1,500 @@ +// 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::collections::BTreeMap; +use std::io::Write; + +use base64::{Engine, engine::general_purpose::STANDARD}; + +use iggy_common::{HeaderKey, HeaderValue}; +use iggy_connector_sdk::{Error, ProducedMessage}; +use lapin::{ + message::Delivery, + types::{AMQPValue, FieldTable}, +}; +use serde_json::{Map, Value, json}; +use tracing::warn; + +pub fn encode_delivery(delivery: &Delivery) -> Result<Vec<u8>, serde_json::Error> { + let mut properties = Map::new(); + for (name, value) in [ Review Comment: warning: `encode_delivery` drops the publisher timestamp and message type even with `include_metadata=true`, so consumers lose event time and type. preserve every present AMQP basic property in the metadata envelope. ########## core/connectors/sources/rabbitmq_source/src/lib.rs: ########## @@ -0,0 +1,919 @@ +// 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. + +pub mod config; +pub mod driver; +pub mod message; + +pub mod batch { + use iggy_connector_sdk::{Error, ProducedMessage, ProducedMessages, Schema}; + use lapin::{Acker, options::BasicAckOptions}; + + pub struct PendingBatch { + schema: Schema, + deliveries: Vec<(ProducedMessage, Acker)>, + in_flight: bool, + } + + impl PendingBatch { + pub fn new(schema: Schema, deliveries: Vec<(ProducedMessage, Acker)>) -> Self { + Self { + schema, + deliveries, + in_flight: false, + } + } + + pub fn next_attempt(&mut self) -> Option<ProducedMessages> { + if self.in_flight { + return None; + } + self.in_flight = true; + // Keep the originals and their ackers for replay if the runtime returns a NACK. + Some(ProducedMessages { + schema: self.schema, + messages: self + .deliveries + .iter() + .map(|(message, _)| ProducedMessage { + id: message.id, + checksum: message.checksum, + timestamp: message.timestamp, + origin_timestamp: message.origin_timestamp, + headers: message.headers.clone(), + payload: message.payload.clone(), + }) + .collect(), + state: None, Review Comment: critical: with `exactly_once=true`, the source reuses producer sequences after restart and acknowledges new messages that iggy discards as duplicates. reject this mode until the source persists and restores replayable batches. ########## core/connectors/sources/rabbitmq_source/src/lib.rs: ########## @@ -0,0 +1,919 @@ +// 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. + +pub mod config; +pub mod driver; +pub mod message; + +pub mod batch { + use iggy_connector_sdk::{Error, ProducedMessage, ProducedMessages, Schema}; + use lapin::{Acker, options::BasicAckOptions}; + + pub struct PendingBatch { + schema: Schema, + deliveries: Vec<(ProducedMessage, Acker)>, + in_flight: bool, + } + + impl PendingBatch { + pub fn new(schema: Schema, deliveries: Vec<(ProducedMessage, Acker)>) -> Self { + Self { + schema, + deliveries, + in_flight: false, + } + } + + pub fn next_attempt(&mut self) -> Option<ProducedMessages> { + if self.in_flight { + return None; + } + self.in_flight = true; + // Keep the originals and their ackers for replay if the runtime returns a NACK. + Some(ProducedMessages { + schema: self.schema, + messages: self + .deliveries + .iter() + .map(|(message, _)| ProducedMessage { + id: message.id, + checksum: message.checksum, + timestamp: message.timestamp, + origin_timestamp: message.origin_timestamp, + headers: message.headers.clone(), + payload: message.payload.clone(), + }) + .collect(), + state: None, + }) + } + + pub fn nack(&mut self) { + self.in_flight = false; + } + + pub async fn acknowledge(&self) -> Result<(), Error> { + for (_, acker) in &self.deliveries { + match acker.ack(BasicAckOptions::default()).await { Review Comment: warning: `PendingBatch` submits and awaits acknowledgements separately for every delivery in the batch. acknowledge the last accepted delivery with `multiple: true` after the batch succeeds. ########## core/connectors/sources/rabbitmq_source/src/message.rs: ########## @@ -0,0 +1,500 @@ +// 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::collections::BTreeMap; +use std::io::Write; + +use base64::{Engine, engine::general_purpose::STANDARD}; + +use iggy_common::{HeaderKey, HeaderValue}; +use iggy_connector_sdk::{Error, ProducedMessage}; +use lapin::{ + message::Delivery, + types::{AMQPValue, FieldTable}, +}; +use serde_json::{Map, Value, json}; +use tracing::warn; + +pub fn encode_delivery(delivery: &Delivery) -> Result<Vec<u8>, serde_json::Error> { + let mut properties = Map::new(); + for (name, value) in [ + ("message_id", delivery.properties.message_id()), + ("content_type", delivery.properties.content_type()), + ("content_encoding", delivery.properties.content_encoding()), + ("correlation_id", delivery.properties.correlation_id()), + ("reply_to", delivery.properties.reply_to()), + ] { + if let Some(value) = value { + properties.insert(name.to_owned(), Value::String(value.as_str().to_owned())); + } + } + let headers = delivery + .properties + .headers() + .as_ref() + .map(encode_table) + .unwrap_or_else(|| Value::Object(Map::new())); + let mut output = Vec::new(); + output.extend_from_slice(b"{\"body_base64\":\""); + { + let mut encoder = base64::write::EncoderWriter::new(&mut output, &STANDARD); + encoder + .write_all(&delivery.data) + .map_err(serde_json::Error::io)?; + encoder.finish().map_err(serde_json::Error::io)?; + } + output.extend_from_slice(b"\",\"delivery\":"); + serde_json::to_writer( + &mut output, + &json!({ + "exchange": delivery.exchange.as_str(), + "routing_key": delivery.routing_key.as_str(), + "redelivered": delivery.redelivered, + }), + )?; + output.extend_from_slice(b",\"headers\":"); + serde_json::to_writer(&mut output, &headers)?; + output.extend_from_slice(b",\"properties\":"); + serde_json::to_writer(&mut output, &properties)?; + output.extend_from_slice(b",\"version\":1}"); + Ok(output) +} + +pub fn encode_table(table: &FieldTable) -> Value { + let fields: Map<String, Value> = table + .inner() + .iter() + .map(|(key, value)| (key.as_str().to_owned(), encode_value(value))) + .collect(); + Value::Object(fields) +} + +pub fn encode_value(value: &AMQPValue) -> Value { + match value { + AMQPValue::Boolean(value) => json!({"type": "boolean", "value": value}), + AMQPValue::ShortShortInt(value) => { + json!({"type": "short_short_int", "value": value.to_string()}) + } + AMQPValue::ShortShortUInt(value) => { + json!({"type": "short_short_uint", "value": value.to_string()}) + } + AMQPValue::ShortInt(value) => json!({"type": "short_int", "value": value.to_string()}), + AMQPValue::ShortUInt(value) => json!({"type": "short_uint", "value": value.to_string()}), + AMQPValue::LongInt(value) => json!({"type": "long_int", "value": value.to_string()}), + AMQPValue::LongUInt(value) => json!({"type": "long_uint", "value": value.to_string()}), + AMQPValue::LongLongInt(value) => { + json!({"type": "long_long_int", "value": value.to_string()}) + } + AMQPValue::Float(value) => { + json!({"type": "float", "value": if value.is_finite() { value.to_string() } else { format_float(*value as f64) }}) + } + AMQPValue::Double(value) => json!({"type": "double", "value": format_float(*value)}), + AMQPValue::DecimalValue(value) => { + json!({"type": "decimal_value", "value": {"scale": value.scale, "value": value.value.to_string()}}) + } + AMQPValue::ShortString(value) => json!({"type": "short_string", "value": value.as_str()}), + AMQPValue::LongString(value) => match std::str::from_utf8(value.as_bytes()) { + Ok(text) => json!({"type": "long_string", "value": text}), + Err(_) => { + json!({"type": "long_string", "value_base64": STANDARD.encode(value.as_bytes())}) + } + }, + AMQPValue::FieldArray(values) => { + json!({"type": "field_array", "value": values.as_slice().iter().map(encode_value).collect::<Vec<_>>()}) + } + AMQPValue::Timestamp(value) => json!({"type": "timestamp", "value": value.to_string()}), + AMQPValue::FieldTable(value) => { + json!({"type": "field_table", "value": encode_table(value)}) + } + AMQPValue::ByteArray(value) => { + json!({"type": "byte_array", "value_base64": STANDARD.encode(value.as_slice())}) + } + AMQPValue::Void => json!({"type": "void", "value": null}), + } +} + +fn format_float(value: f64) -> String { + if value.is_nan() { + "NaN".to_owned() + } else if value == f64::INFINITY { + "Infinity".to_owned() + } else if value == f64::NEG_INFINITY { + "-Infinity".to_owned() + } else { + value.to_string() + } +} + +const MAX_PAYLOAD_BYTES: usize = 64_000_000; +const MAX_HEADER_BYTES: usize = 100_000; + +pub fn prepare_delivery( + delivery: &mut Delivery, + include_metadata: bool, + connector_id: u32, +) -> Result<Option<ProducedMessage>, Error> { + if unsupported_body(delivery.data.len(), include_metadata) { + return Ok(None); + } + let (payload, headers) = if include_metadata { + let payload = + encode_delivery(delivery).map_err(|error| Error::Serialization(error.to_string()))?; + if exceeds_payload_limit(payload.len()) { + return Ok(None); + } + (payload, None) + } else { + let headers = raw_headers(delivery, connector_id)?; + (std::mem::take(&mut delivery.data), Some(headers)) + }; + Ok(Some(ProducedMessage { + id: None, + checksum: None, + timestamp: None, + origin_timestamp: None, + headers, + payload, + })) +} + +fn unsupported_body(size: usize, include_metadata: bool) -> bool { + exceeds_payload_limit(size) || (!include_metadata && size == 0) +} + +fn exceeds_payload_limit(size: usize) -> bool { + size > MAX_PAYLOAD_BYTES +} + +fn raw_headers( + delivery: &Delivery, + connector_id: u32, +) -> Result<BTreeMap<HeaderKey, HeaderValue>, Error> { + let mut headers = BTreeMap::new(); + let mut size = 0; + for (key, value) in [ + ("amqp.exchange", delivery.exchange.as_str()), + ("amqp.routing_key", delivery.routing_key.as_str()), + ] { + if !value.is_empty() { + insert_string(&mut headers, &mut size, key, value)?; + } + } + let key = HeaderKey::try_from("amqp.redelivered") + .map_err(|error| Error::Serialization(error.to_string()))?; + size += 10 + key.as_bytes().len() + 1; + headers.insert(key, HeaderValue::from(delivery.redelivered)); + if let Some(value) = delivery.properties.message_id() { + let key = "amqp.message_id"; + let value = value.as_str(); + if value.is_empty() + || value.len() > 255 + || size + 10 + key.len() + value.len() > MAX_HEADER_BYTES Review Comment: simplification: the four validated headers cannot reach this size budget. remove the size accumulator and its helper parameter while retaining each field's length validation. ########## core/integration/tests/connectors/rabbitmq/rabbitmq_source.rs: ########## @@ -0,0 +1,423 @@ +// 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::time::Duration; + +use crate::connectors::fixtures::{RabbitMqSourceFixture, RabbitMqSourceMetadataFixture}; +use iggy_common::poll_messages::DEFAULT_PARTITION_ID; +use iggy_common::{ + CommitPolicy, Consumer, Identifier, IggyMessage, MessageClient, PollOptions, PollingStrategy, +}; +use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStats, ConnectorStatus}; +use integration::harness::{TestHarness, seeds}; +use integration::iggy_harness; +use lapin::{ + BasicProperties, + types::{AMQPValue, FieldTable}, +}; +use reqwest::Client; +use serde_json::Value; +use tokio::time::{sleep, timeout}; + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/source.toml")), + seed = seeds::connector_stream +)] +async fn rabbitmq_source_preserves_raw_payload( + harness: &TestHarness, + fixture: RabbitMqSourceFixture, +) { + wait_for_running(harness).await; + fixture + .publish(&[0, 255, 1], BasicProperties::default()) + .await + .expect("RabbitMQ publish must succeed"); + let message = next_message(harness).await; + assert_eq!(message.payload.as_ref(), &[0, 255, 1]); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/source.toml")), + seed = seeds::connector_stream +)] +async fn given_supported_delivery_when_iggy_accepts_batch_should_acknowledge_rabbitmq( + harness: &TestHarness, + fixture: RabbitMqSourceFixture, +) { + wait_for_running(harness).await; + fixture + .publish(b"ack-after-iggy-accepts", BasicProperties::default()) + .await + .expect("RabbitMQ publish must succeed"); + + let message = next_message(harness).await; + assert_eq!(message.payload.as_ref(), b"ack-after-iggy-accepts"); + + timeout(Duration::from_secs(10), async { + loop { + if !fixture + .get_ready_message() + .await + .expect("RabbitMQ queue inspection must succeed") + { + break; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("accepted delivery must leave the RabbitMQ ready queue"); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/source.toml")), + seed = seeds::connector_stream +)] +async fn rabbitmq_source_reconnects_after_broker_outage( + harness: &TestHarness, + fixture: RabbitMqSourceFixture, +) { + wait_for_running(harness).await; + fixture + .pause_broker() + .await + .expect("RabbitMQ broker should pause"); + sleep(Duration::from_secs(5)).await; Review Comment: warning: this test resumes the broker without proving that the source disconnected, so the existing connection can satisfy the assertion. force the source connection closed and require a new consumer before the final publish. ########## core/integration/tests/connectors/rabbitmq/rabbitmq_source.rs: ########## @@ -0,0 +1,423 @@ +// 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::time::Duration; + +use crate::connectors::fixtures::{RabbitMqSourceFixture, RabbitMqSourceMetadataFixture}; +use iggy_common::poll_messages::DEFAULT_PARTITION_ID; +use iggy_common::{ + CommitPolicy, Consumer, Identifier, IggyMessage, MessageClient, PollOptions, PollingStrategy, +}; +use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStats, ConnectorStatus}; +use integration::harness::{TestHarness, seeds}; +use integration::iggy_harness; +use lapin::{ + BasicProperties, + types::{AMQPValue, FieldTable}, +}; +use reqwest::Client; +use serde_json::Value; +use tokio::time::{sleep, timeout}; + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/source.toml")), + seed = seeds::connector_stream +)] +async fn rabbitmq_source_preserves_raw_payload( + harness: &TestHarness, + fixture: RabbitMqSourceFixture, +) { + wait_for_running(harness).await; + fixture + .publish(&[0, 255, 1], BasicProperties::default()) + .await + .expect("RabbitMQ publish must succeed"); + let message = next_message(harness).await; + assert_eq!(message.payload.as_ref(), &[0, 255, 1]); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/source.toml")), + seed = seeds::connector_stream +)] +async fn given_supported_delivery_when_iggy_accepts_batch_should_acknowledge_rabbitmq( + harness: &TestHarness, + fixture: RabbitMqSourceFixture, +) { + wait_for_running(harness).await; + fixture + .publish(b"ack-after-iggy-accepts", BasicProperties::default()) + .await + .expect("RabbitMQ publish must succeed"); + + let message = next_message(harness).await; + assert_eq!(message.payload.as_ref(), b"ack-after-iggy-accepts"); + + timeout(Duration::from_secs(10), async { + loop { + if !fixture + .get_ready_message() Review Comment: warning: this test passes with deliveries still unacknowledged because `basic_get` cannot read messages owned by the source consumer. close the source connection after iggy delivery and assert that rabbitmq contains no redelivery. -- 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]
