From 2cc73bec41496f07ea024cf439b7b4e2ed575f1c Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 11:15:34 -0400 Subject: [PATCH 1/2] test(eventhubs): cover producer routing and payload edges Add tests/eventhubs_producer_routing.rs. The file covers the partition key routing of a batch, the zero length body, the large body, the oversized single event, the AMQP value body, the AMQP sequence body, and the binary application property. Each test captures the tail sequence number of every partition that it reads, then reads strictly after that boundary. Every event carries a per run marker, and every receive side assertion filters on it. No assertion depends on a partition tail staying still, so the tests need no lock against each other, and only one receiver is open at a time. The eight live tests are unproven until a live run. The test macro marks them as ignored when the compile time mode is below Live, so they neither pass nor fail here. The two plain unit tests carry a mutation proof: a contains_key marker check makes tagged_label accept another run, and a constant fill or a 256 byte fill makes large_body fail its pattern assertions. The file needs no Cargo.toml change, so it cannot conflict with PR 5077, which adds the tokio sync feature to the same manifest. --- .../tests/eventhubs_producer_routing.rs | 922 ++++++++++++++++++ 1 file changed, 922 insertions(+) create mode 100644 sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs diff --git a/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs new file mode 100644 index 00000000000..dc4f7500eac --- /dev/null +++ b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs @@ -0,0 +1,922 @@ +// Copyright (c) Microsoft Corporation. All Rights reserved +// Licensed under the MIT license. + +//! Live tests for producer routing and for the payload edges of an event. +//! +//! Each test captures the tail sequence number of a partition before it sends, +//! then reads strictly after that boundary. Every event carries a per-run +//! marker, so a test asserts on its own events and ignores the foreign traffic +//! that shares the partition. No assertion depends on a partition tail staying +//! still, so these tests need no lock against each other. + +use azure_core_amqp::{message::AmqpMessageBody, AmqpList}; +use azure_core_test::{recorded, TestContext}; +use azure_messaging_eventhubs::{ + models::{AmqpMessage, AmqpSimpleValue, AmqpValue, EventData, ReceivedEventData}, + ConsumerClient, EventDataBatchOptions, EventReceiver, OpenReceiverOptions, ProducerClient, + SendEventOptions, StartLocation, StartPosition, +}; +use futures::stream::StreamExt; +use std::{collections::HashMap, env, error::Error, time::Duration}; +use tracing::info; + +const RUN_MARKER_KEY: &str = "producer-routing-run"; +const EVENT_LABEL_KEY: &str = "producer-routing-label"; +const ROUTED_EVENT_COUNT: usize = 5; + +// Event Hubs caps a single event by tier: 256 KB on Basic, 1 MB on Standard, +// 1 MB on Premium, and 20 MB on Dedicated. +// https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-quotas +// 100_000 bytes is safe on every tier. The .NET test +// ProducerCanSendSingleLargeEventInASet uses new byte[100000] with the comment +// "Actual limit is 1046520 for a single event". +const LARGE_BODY_LEN: usize = 100_000; +const OVERSIZED_BODY_LEN: usize = 2 * 1024 * 1024; +const SEQUENCE_INT_VALUE: i32 = 1_234_567_890; +const BINARY_PROPERTY_KEY: &str = "producer-routing-binary"; +const BINARY_PROPERTY_VALUE: [u8; 4] = [0x12, 0x34, 0x56, 0x78]; + +// A sweep pays this deadline once for each partition that holds no marked +// event. A namespace with 32 partitions can cost about 10 minutes in the worst +// ordering. +const PARTITION_SWEEP_DEADLINE: Duration = Duration::from_secs(20); +const READ_DEADLINE: Duration = Duration::from_secs(60); +const OVERSIZED_SEND_DEADLINE: Duration = Duration::from_secs(45); + +/// Returns the label of an event when the event belongs to this run. +/// +/// The function takes the property map and not the event, so a plain unit test +/// can build the input. +fn tagged_label( + properties: Option<&HashMap>, + run_marker: &str, +) -> Option { + let properties = properties?; + match properties.get(RUN_MARKER_KEY) { + Some(AmqpSimpleValue::String(marker)) if marker == run_marker => {} + _ => return None, + } + match properties.get(EVENT_LABEL_KEY) { + Some(AmqpSimpleValue::String(label)) => Some(label.clone()), + _ => None, + } +} + +/// Builds a body of `len` bytes that holds a repeating pattern. +/// +/// 251 is a prime number, so the pattern never aligns with a 256 byte block or +/// with any other power of two block. +fn large_body(len: usize) -> Vec { + (0..len).map(|index| (index % 251) as u8).collect() +} + +/// Opens a producer for `test_name`. +async fn open_producer( + ctx: &TestContext, + test_name: &str, +) -> Result> { + let host = env::var("EVENTHUBS_HOST")?; + let eventhub = env::var("EVENTHUB_NAME")?; + let credential = ctx.recording().credential(); + Ok(ProducerClient::builder() + .with_application_id(test_name.to_string()) + .open(host.as_str(), eventhub.as_str(), credential) + .await?) +} + +/// Opens a consumer for `test_name`. +async fn open_consumer( + ctx: &TestContext, + test_name: &str, +) -> Result> { + let host = env::var("EVENTHUBS_HOST")?; + let eventhub = env::var("EVENTHUB_NAME")?; + let credential = ctx.recording().credential(); + Ok(ConsumerClient::builder() + .with_application_id(test_name.to_string()) + .open(host.as_str(), eventhub, credential) + .await?) +} + +/// Reads until `want` events of this run arrive or `deadline` expires. +async fn read_run_events( + receiver: &EventReceiver, + run_marker: &str, + want: usize, + deadline: Duration, +) -> Vec { + let mut found: Vec = Vec::new(); + let mut stream = receiver.stream_events(); + + // The stream does not end on its own, so the deadline is the only exit. + let _ = tokio::time::timeout(deadline, async { + while let Some(event) = stream.next().await { + match event { + Ok(event) => { + if tagged_label(event.event_data().properties(), run_marker).is_some() { + found.push(event); + } + if found.len() >= want { + break; + } + } + Err(err) => { + info!("run {run_marker}: the stream failed, stop reading. {err:?}"); + break; + } + } + } + }) + .await; + + found +} + +/// Reads the events of this run from one partition, starting after +/// `from_sequence`. Exactly one receiver is open at any moment. +async fn read_run_events_on_partition( + consumer: &ConsumerClient, + partition: &str, + from_sequence: i64, + run_marker: &str, + want: usize, + deadline: Duration, +) -> Result, Box> { + let receiver = consumer + .open_receiver_on_partition( + partition.to_string(), + Some(OpenReceiverOptions { + start_position: Some(StartPosition { + location: StartLocation::SequenceNumber(from_sequence), + ..Default::default() + }), + ..Default::default() + }), + ) + .await?; + let events = read_run_events(&receiver, run_marker, want, deadline).await; + receiver.close().await?; + Ok(events) +} + +/// Reads the last sequence number of every partition. A test reads strictly +/// after these numbers, so a foreign event that came before is never read. +async fn capture_tails( + producer: &ProducerClient, + partitions: &[String], +) -> Result, Box> { + let mut tails = HashMap::new(); + for partition in partitions { + let properties = producer.get_partition_properties(partition).await?; + tails.insert(partition.clone(), properties.last_enqueued_sequence_number); + } + Ok(tails) +} + +/// Builds an event that carries the run marker and its own label. +fn marked_event(run_marker: &str, label: &str, body: impl Into>) -> EventData { + EventData::builder() + .with_body(body) + .add_property(RUN_MARKER_KEY.to_string(), run_marker.to_string()) + .add_property(EVENT_LABEL_KEY.to_string(), label.to_string()) + .build() +} + +/// Sends one batch with `key` as the partition key. The batch names no +/// partition id, so the service picks the partition from the key. +async fn send_keyed_batch( + producer: &ProducerClient, + key: &str, + run_marker: &str, + labels: &[String], +) -> Result<(), Box> { + let batch = producer + .create_batch(Some(EventDataBatchOptions { + partition_key: Some(key.to_string()), + partition_id: None, + ..Default::default() + })) + .await?; + for label in labels { + assert!( + batch.try_add_event_data(marked_event(run_marker, label, label.clone()), None)?, + "run {run_marker}: event {label} did not fit in the batch for key {key}" + ); + } + producer.send_batch(batch, None).await?; + Ok(()) +} + +/// Returns the first partition in `order` that holds every marked event of one +/// batch of this run. +async fn sweep_for_batch( + consumer: &ConsumerClient, + order: &[String], + tails: &HashMap, + run_marker: &str, +) -> Result, Box> { + for partition in order { + let events = read_run_events_on_partition( + consumer, + partition, + tails[partition], + run_marker, + ROUTED_EVENT_COUNT, + PARTITION_SWEEP_DEADLINE, + ) + .await?; + if events.len() >= ROUTED_EVENT_COUNT { + return Ok(Some(partition.clone())); + } + } + Ok(None) +} + +/// A batch that carries a partition key must land on exactly one partition, and +/// every event of that batch must arrive with the key. +/// +/// The test never asserts a specific partition id, and it never asserts that a +/// different key lands on a different partition. The hash is stable only for a +/// fixed partition count, and a hash over 4 partitions can put two keys on the +/// same partition. +#[recorded::test(live)] +async fn partition_key_routes_batch_to_one_partition( + ctx: TestContext, +) -> Result<(), Box> { + const TEST_NAME: &str = "partition_key_routes_batch_to_one_partition"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + + let key = format!("routing-{}", azure_core::Uuid::new_v4()); + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let partitions = producer.get_eventhub_properties().await?.partition_ids; + assert!( + partitions.len() > 1, + "run {run_marker}: the Event Hub reports {} partition(s), so this test cannot tell one \ + partition from another", + partitions.len() + ); + + let tails = capture_tails(&producer, &partitions).await?; + let labels: Vec = (0..ROUTED_EVENT_COUNT) + .map(|index| format!("routed-{index}")) + .collect(); + send_keyed_batch(&producer, &key, &run_marker, &labels).await?; + + let mut carriers: Vec<(String, usize)> = Vec::new(); + let mut received_keys: Vec> = Vec::new(); + for partition in &partitions { + let events = read_run_events_on_partition( + &consumer, + partition, + tails[partition], + &run_marker, + ROUTED_EVENT_COUNT, + PARTITION_SWEEP_DEADLINE, + ) + .await?; + if events.is_empty() { + continue; + } + for event in &events { + received_keys.push(event.partition_key().clone()); + } + let count = events.len(); + carriers.push((partition.clone(), count)); + // The test sent exactly ROUTED_EVENT_COUNT events, so once one partition + // holds all of them no other partition can hold one. Stop the sweep and + // save the deadline that each remaining partition would cost. + if count >= ROUTED_EVENT_COUNT { + break; + } + } + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + carriers.len(), + 1, + "run {run_marker}: key {key} put events on {} partition(s) of {}, and a partition key must \ + pick exactly one. Each partition read had a {PARTITION_SWEEP_DEADLINE:?} deadline. The \ + partitions that carried events were {carriers:?}.", + carriers.len(), + partitions.len() + ); + let (carrier, count) = &carriers[0]; + assert_eq!( + *count, ROUTED_EVENT_COUNT, + "run {run_marker}: partition {carrier} gave {count} of the {ROUTED_EVENT_COUNT} events of \ + key {key} within {PARTITION_SWEEP_DEADLINE:?}" + ); + for partition_key in &received_keys { + assert_eq!( + partition_key.as_deref(), + Some(key.as_str()), + "run {run_marker}: an event on partition {carrier} arrived with the partition key \ + {partition_key:?}, and every event of this batch must carry {key}" + ); + } + Ok(()) +} + +/// Two batches that carry the same partition key must land on the same +/// partition. +#[recorded::test(live)] +async fn same_partition_key_routes_to_same_partition( + ctx: TestContext, +) -> Result<(), Box> { + const TEST_NAME: &str = "same_partition_key_routes_to_same_partition"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + + let key = format!("routing-{}", azure_core::Uuid::new_v4()); + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let partitions = producer.get_eventhub_properties().await?.partition_ids; + assert!( + partitions.len() > 1, + "run {run_marker}: the Event Hub reports {} partition(s), so this test cannot tell one \ + partition from another", + partitions.len() + ); + + let tails_a = capture_tails(&producer, &partitions).await?; + let labels_a: Vec = (0..ROUTED_EVENT_COUNT) + .map(|index| format!("a-{index}")) + .collect(); + send_keyed_batch(&producer, &key, &run_marker, &labels_a).await?; + let partition_a = sweep_for_batch(&consumer, &partitions, &tails_a, &run_marker) + .await? + .ok_or_else(|| { + format!( + "run {run_marker}: no partition of the {} gave all {ROUTED_EVENT_COUNT} events of \ + batch a for key {key}. Each partition read had a {PARTITION_SWEEP_DEADLINE:?} \ + deadline.", + partitions.len() + ) + })?; + + // The second capture is the boundary for batch b, so every marked event that + // the second sweep reads is a batch b event. + let tails_b = capture_tails(&producer, &partitions).await?; + let labels_b: Vec = (0..ROUTED_EVENT_COUNT) + .map(|index| format!("b-{index}")) + .collect(); + send_keyed_batch(&producer, &key, &run_marker, &labels_b).await?; + + // Read partition_a first. The passing case then costs one fast read, and the + // failing case still names where batch b landed. + let mut order: Vec = vec![partition_a.clone()]; + order.extend( + partitions + .iter() + .filter(|partition| **partition != partition_a) + .cloned(), + ); + let partition_b = sweep_for_batch(&consumer, &order, &tails_b, &run_marker).await?; + + consumer.close().await?; + producer.close().await?; + + let partition_b = partition_b.ok_or_else(|| { + format!( + "run {run_marker}: no partition of the {} gave all {ROUTED_EVENT_COUNT} events of \ + batch b for key {key}, and batch a landed on partition {partition_a}. Each partition \ + read had a {PARTITION_SWEEP_DEADLINE:?} deadline.", + partitions.len() + ) + })?; + assert_eq!( + partition_b, partition_a, + "run {run_marker}: key {key} sent batch a to partition {partition_a} and batch b to \ + partition {partition_b}. One key must always pick the same partition. Each partition read \ + had a {PARTITION_SWEEP_DEADLINE:?} deadline." + ); + Ok(()) +} + +/// A zero length body must arrive as an empty slice. +#[recorded::test(live)] +async fn zero_length_body_round_trips_as_empty_slice( + ctx: TestContext, +) -> Result<(), Box> { + const TEST_NAME: &str = "zero_length_body_round_trips_as_empty_slice"; + const PARTITION: &str = "0"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let tail = producer + .get_partition_properties(PARTITION) + .await? + .last_enqueued_sequence_number; + + // The explicit empty Vec makes a zero length Data section, which matches the + // .NET Array.Empty() case. A builder with no with_body call makes an + // absent body section, a different wire shape, so do not omit the call. + producer + .send_event( + marked_event(&run_marker, "empty-body", Vec::::new()), + Some(SendEventOptions { + partition_id: Some(PARTITION.to_string()), + }), + ) + .await?; + + let events = + read_run_events_on_partition(&consumer, PARTITION, tail, &run_marker, 1, READ_DEADLINE) + .await?; + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + events.len(), + 1, + "run {run_marker}: partition {PARTITION} gave {} marked event(s) within {READ_DEADLINE:?} \ + after sequence number {tail}, and the test wanted 1", + events.len() + ); + assert_eq!( + events[0].event_data().body(), + Some(&[][..]), + "run {run_marker}: a zero length body must arrive as an empty slice and not as an absent \ + body. Partition {PARTITION}, after sequence number {tail}, deadline {READ_DEADLINE:?}." + ); + Ok(()) +} + +/// A large body must arrive byte for byte. +#[recorded::test(live)] +async fn large_event_body_round_trips_intact(ctx: TestContext) -> Result<(), Box> { + const TEST_NAME: &str = "large_event_body_round_trips_intact"; + const PARTITION: &str = "0"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + let body = large_body(LARGE_BODY_LEN); + + let tail = producer + .get_partition_properties(PARTITION) + .await? + .last_enqueued_sequence_number; + + // The batch names no max_size_in_bytes, so the sender link maximum applies + // and the batch accepts an event of the size the service allows. + let batch = producer + .create_batch(Some(EventDataBatchOptions { + partition_id: Some(PARTITION.to_string()), + ..Default::default() + })) + .await?; + assert!( + batch.try_add_event_data(marked_event(&run_marker, "large-body", body.clone()), None)?, + "run {run_marker}: a {LARGE_BODY_LEN} byte event must fit under the sender link maximum; a \ + smaller Event Hubs tier fails here. Partition {PARTITION}." + ); + producer.send_batch(batch, None).await?; + + let events = + read_run_events_on_partition(&consumer, PARTITION, tail, &run_marker, 1, READ_DEADLINE) + .await?; + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + events.len(), + 1, + "run {run_marker}: partition {PARTITION} gave {} marked event(s) within {READ_DEADLINE:?} \ + after sequence number {tail}, and the test wanted 1", + events.len() + ); + let received = events[0].event_data().body().unwrap_or_else(|| { + panic!( + "run {run_marker}: the large event arrived with no body from partition {PARTITION} \ + within {READ_DEADLINE:?} after sequence number {tail}" + ) + }); + + // Compare the length first and then the first difference. A direct + // assert_eq! of two 100 KB slices prints an unreadable panic. + assert_eq!( + received.len(), + LARGE_BODY_LEN, + "run {run_marker}: partition {PARTITION} gave a body of {} bytes within {READ_DEADLINE:?}, \ + and the test sent {LARGE_BODY_LEN} bytes", + received.len() + ); + let first_diff = received + .iter() + .zip(body.iter()) + .position(|(left, right)| left != right); + assert!( + first_diff.is_none(), + "run {run_marker}: the received body differs from the sent body at index {first_diff:?}. \ + Partition {PARTITION}, deadline {READ_DEADLINE:?}." + ); + Ok(()) +} + +/// A single event above the service cap must not send. +/// +/// The batch half of this behavior lives in the other file: +/// create_batch_honors_max_size_in_bytes and +/// create_batch_rejects_size_above_link_maximum in tests/eventhubs_producer.rs +/// cover the batch refusal. This test covers the send_event path only. +#[recorded::test(live)] +async fn oversized_send_event_does_not_succeed(ctx: TestContext) -> Result<(), Box> { + const TEST_NAME: &str = "oversized_send_event_does_not_succeed"; + const PARTITION: &str = "0"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + let body = large_body(OVERSIZED_BODY_LEN); + + // The test asserts nothing about the error kind or the message text. The + // error module has no size limit variant, and nothing pins how the fe2o3 + // link payload error surfaces. Tighten this to the real error after a live + // run shows the kind. + let outcome = tokio::time::timeout( + OVERSIZED_SEND_DEADLINE, + producer.send_event( + EventData::builder() + .with_body(body) + .add_property(RUN_MARKER_KEY.to_string(), run_marker.clone()) + .build(), + Some(SendEventOptions { + partition_id: Some(PARTITION.to_string()), + }), + ), + ) + .await; + + match outcome { + Err(_elapsed) => info!( + "run {run_marker}: the {OVERSIZED_BODY_LEN} byte send to partition {PARTITION} did not \ + finish within {OVERSIZED_SEND_DEADLINE:?}, so it did not succeed." + ), + Ok(Err(err)) => info!( + "run {run_marker}: the service rejected the {OVERSIZED_BODY_LEN} byte send to \ + partition {PARTITION}. {err:?}" + ), + Ok(Ok(())) => panic!( + "run {run_marker}: a {OVERSIZED_BODY_LEN} byte event reached partition {PARTITION} \ + within {OVERSIZED_SEND_DEADLINE:?}, but the Standard tier caps a single event at \ + about 1 MB" + ), + } + + // The oversized send can already have detached the link, so a close error is + // not a test failure. + let _ = producer.close().await; + Ok(()) +} + +/// An AMQP value body must arrive as a value body, and EventData must report no +/// body for it. +#[recorded::test(live)] +async fn amqp_value_body_round_trips_without_event_data_body( + ctx: TestContext, +) -> Result<(), Box> { + const TEST_NAME: &str = "amqp_value_body_round_trips_without_event_data_body"; + const PARTITION: &str = "0"; + const VALUE_BODY: &str = "producer-routing-value-body"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let tail = producer + .get_partition_properties(PARTITION) + .await? + .last_enqueued_sequence_number; + + let batch = producer + .create_batch(Some(EventDataBatchOptions { + partition_id: Some(PARTITION.to_string()), + ..Default::default() + })) + .await?; + assert!( + batch.try_add_amqp_message( + AmqpMessage::builder() + .with_body(AmqpValue::from(VALUE_BODY)) + .add_application_property(RUN_MARKER_KEY.to_string(), run_marker.clone()) + .add_application_property(EVENT_LABEL_KEY.to_string(), "value-body".to_string()) + .build(), + None, + )?, + "run {run_marker}: the value body message did not fit in the batch for partition \ + {PARTITION}" + ); + producer.send_batch(batch, None).await?; + + let events = + read_run_events_on_partition(&consumer, PARTITION, tail, &run_marker, 1, READ_DEADLINE) + .await?; + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + events.len(), + 1, + "run {run_marker}: partition {PARTITION} gave {} marked event(s) within {READ_DEADLINE:?} \ + after sequence number {tail}, and the test wanted 1", + events.len() + ); + let event = &events[0]; + match &event.raw_amqp_message().body { + AmqpMessageBody::Value(AmqpValue::String(text)) => assert_eq!( + text.as_str(), + VALUE_BODY, + "run {run_marker}: the value body arrived with the wrong text from partition \ + {PARTITION} within {READ_DEADLINE:?}" + ), + other => panic!( + "run {run_marker}: the value body must arrive as AmqpMessageBody::Value(String) from \ + partition {PARTITION} within {READ_DEADLINE:?}, got {other:?}" + ), + } + + // EventData::from_message copies a body only for an AmqpMessageBody::Binary + // that holds exactly one element, so a value body gives None by design. + assert!( + event.event_data().body().is_none(), + "run {run_marker}: a value body must give no EventData body, because only a single element \ + binary body is copied. Partition {PARTITION}, deadline {READ_DEADLINE:?}." + ); + Ok(()) +} + +/// An AMQP sequence body must arrive with its elements in order and with the +/// type of each element intact. +#[recorded::test(live)] +async fn amqp_sequence_body_round_trips_in_order(ctx: TestContext) -> Result<(), Box> { + const TEST_NAME: &str = "amqp_sequence_body_round_trips_in_order"; + const PARTITION: &str = "0"; + const SEQUENCE_TEXT: &str = "sequence-text"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let tail = producer + .get_partition_properties(PARTITION) + .await? + .last_enqueued_sequence_number; + + let batch = producer + .create_batch(Some(EventDataBatchOptions { + partition_id: Some(PARTITION.to_string()), + ..Default::default() + })) + .await?; + // A small integer can go on the wire in one byte and come back in a narrower + // variant, which would make the assertion below flake on the encoding width. + // A value above 2^16 forces a 4 byte integer. + assert!( + batch.try_add_amqp_message( + AmqpMessage::builder() + .with_body(vec![ + AmqpValue::from(SEQUENCE_TEXT), + AmqpValue::from(SEQUENCE_INT_VALUE), + ]) + .add_application_property(RUN_MARKER_KEY.to_string(), run_marker.clone()) + .add_application_property(EVENT_LABEL_KEY.to_string(), "sequence-body".to_string()) + .build(), + None, + )?, + "run {run_marker}: the sequence body message did not fit in the batch for partition \ + {PARTITION}" + ); + producer.send_batch(batch, None).await?; + + let events = + read_run_events_on_partition(&consumer, PARTITION, tail, &run_marker, 1, READ_DEADLINE) + .await?; + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + events.len(), + 1, + "run {run_marker}: partition {PARTITION} gave {} marked event(s) within {READ_DEADLINE:?} \ + after sequence number {tail}, and the test wanted 1", + events.len() + ); + match &events[0].raw_amqp_message().body { + // From> for AmqpMessageBody wraps the values in one + // AmqpList, so the sequence holds exactly one list. + AmqpMessageBody::Sequence(lists) => { + assert_eq!( + lists.len(), + 1, + "run {run_marker}: the sequence body must hold exactly one list from partition \ + {PARTITION} within {READ_DEADLINE:?}, got {} list(s)", + lists.len() + ); + let list: &AmqpList = &lists[0]; + let values = &list.0; + assert_eq!( + values.len(), + 2, + "run {run_marker}: the sequence list must hold the 2 values that the test sent to \ + partition {PARTITION}, got {} within {READ_DEADLINE:?}", + values.len() + ); + assert_eq!( + values[0], + AmqpValue::String(SEQUENCE_TEXT.to_string()), + "run {run_marker}: the first sequence element must arrive as the text that the \ + test sent to partition {PARTITION} within {READ_DEADLINE:?}" + ); + match &values[1] { + AmqpValue::Int(value) => assert_eq!( + *value, SEQUENCE_INT_VALUE, + "run {run_marker}: the second sequence element arrived with the wrong integer \ + from partition {PARTITION} within {READ_DEADLINE:?}" + ), + other => panic!( + "run {run_marker}: the second sequence element must arrive as an integer from \ + partition {PARTITION} within {READ_DEADLINE:?}, got {other:?}" + ), + } + } + other => panic!( + "run {run_marker}: the sequence body must arrive as AmqpMessageBody::Sequence from \ + partition {PARTITION} within {READ_DEADLINE:?}, got {other:?}" + ), + } + Ok(()) +} + +/// A binary application property must arrive with the exact bytes. +#[recorded::test(live)] +async fn binary_application_property_round_trips(ctx: TestContext) -> Result<(), Box> { + const TEST_NAME: &str = "binary_application_property_round_trips"; + const PARTITION: &str = "0"; + + let producer = open_producer(&ctx, TEST_NAME).await?; + let consumer = open_consumer(&ctx, TEST_NAME).await?; + let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); + + let tail = producer + .get_partition_properties(PARTITION) + .await? + .last_enqueued_sequence_number; + + producer + .send_event( + EventData::builder() + .with_body("binary-property") + .add_property(RUN_MARKER_KEY.to_string(), run_marker.clone()) + .add_property(EVENT_LABEL_KEY.to_string(), "binary-property".to_string()) + .add_property( + BINARY_PROPERTY_KEY.to_string(), + AmqpSimpleValue::Binary(BINARY_PROPERTY_VALUE.to_vec()), + ) + .build(), + Some(SendEventOptions { + partition_id: Some(PARTITION.to_string()), + }), + ) + .await?; + + let events = + read_run_events_on_partition(&consumer, PARTITION, tail, &run_marker, 1, READ_DEADLINE) + .await?; + + consumer.close().await?; + producer.close().await?; + + assert_eq!( + events.len(), + 1, + "run {run_marker}: partition {PARTITION} gave {} marked event(s) within {READ_DEADLINE:?} \ + after sequence number {tail}, and the test wanted 1", + events.len() + ); + let properties = events[0].event_data().properties().unwrap_or_else(|| { + panic!( + "run {run_marker}: the event arrived with no application properties from partition \ + {PARTITION} within {READ_DEADLINE:?}" + ) + }); + assert_eq!( + properties.get(BINARY_PROPERTY_KEY), + Some(&AmqpSimpleValue::Binary(BINARY_PROPERTY_VALUE.to_vec())), + "run {run_marker}: the binary application property must arrive with the exact bytes from \ + partition {PARTITION} within {READ_DEADLINE:?}" + ); + Ok(()) +} + +/// `tagged_label` must give a label only for an event of this run. +#[test] +fn tagged_label_matches_only_the_run_marker() { + assert!( + tagged_label(None, "run-1").is_none(), + "an event with no application properties cannot belong to a run" + ); + + let mut matching = HashMap::new(); + matching.insert( + RUN_MARKER_KEY.to_string(), + AmqpSimpleValue::String("run-1".to_string()), + ); + matching.insert( + EVENT_LABEL_KEY.to_string(), + AmqpSimpleValue::String("x".to_string()), + ); + assert_eq!( + tagged_label(Some(&matching), "run-1"), + Some("x".to_string()), + "the marker of this run with a string label must give the label" + ); + + let mut other_run = HashMap::new(); + other_run.insert( + RUN_MARKER_KEY.to_string(), + AmqpSimpleValue::String("run-2".to_string()), + ); + other_run.insert( + EVENT_LABEL_KEY.to_string(), + AmqpSimpleValue::String("x".to_string()), + ); + assert!( + tagged_label(Some(&other_run), "run-1").is_none(), + "the marker of another run must give no label" + ); + + let mut integer_marker = HashMap::new(); + integer_marker.insert(RUN_MARKER_KEY.to_string(), AmqpSimpleValue::Int(1)); + integer_marker.insert( + EVENT_LABEL_KEY.to_string(), + AmqpSimpleValue::String("x".to_string()), + ); + assert!( + tagged_label(Some(&integer_marker), "run-1").is_none(), + "a marker that is not a string must give no label" + ); + + let mut no_label = HashMap::new(); + no_label.insert( + RUN_MARKER_KEY.to_string(), + AmqpSimpleValue::String("run-1".to_string()), + ); + assert!( + tagged_label(Some(&no_label), "run-1").is_none(), + "the marker of this run with no label key must give no label" + ); + + let mut integer_label = HashMap::new(); + integer_label.insert( + RUN_MARKER_KEY.to_string(), + AmqpSimpleValue::String("run-1".to_string()), + ); + integer_label.insert(EVENT_LABEL_KEY.to_string(), AmqpSimpleValue::Int(7)); + assert!( + tagged_label(Some(&integer_label), "run-1").is_none(), + "a label that is not a string must give no label" + ); +} + +/// `large_body` must give the length that the caller asked for, and the fill +/// must vary. +#[test] +fn large_body_fills_a_non_constant_pattern() { + assert_eq!( + large_body(LARGE_BODY_LEN).len(), + LARGE_BODY_LEN, + "the helper must give the length that the caller asked for" + ); + assert!( + large_body(0).is_empty(), + "a zero length request must give an empty body" + ); + + let body = large_body(512); + assert_ne!( + body[0], body[1], + "the fill must vary so a zero filled buffer cannot pass by accident" + ); + assert!( + body.iter() + .enumerate() + .all(|(index, byte)| *byte == (index % 251) as u8), + "every byte must follow the 251 byte pattern" + ); + assert_ne!( + body[0], body[256], + "the fill period must not align with a power of two block" + ); +} From 6bd09de1cd9f4aa9e56f2ba8ec6aa8b5baf53f21 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 12:36:25 -0400 Subject: [PATCH 2/2] test(eventhubs): assert the batch link maximum, not a single send A live run showed that the single event path does not enforce the negotiated AMQP link maximum. The sender link reported `max_allowed: 1048576` on a Standard namespace, and `send_event` still accepted a 2 MiB event, returned `Ok(())`, and moved the partition tail. The batch path refuses the same event, so the old assertion could never pass against a real broker. Assert the batch refusal instead, which is also what the .NET `ProducerCannotSendSetLargerThanMaximumSize` asserts. Read the maximum from the `InvalidBatchSize` error rather than hardcoding it, so the test holds on every tier. The single event divergence needs its own issue. --- .../tests/eventhubs_producer_routing.rs | 107 ++++++++++-------- 1 file changed, 60 insertions(+), 47 deletions(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs index dc4f7500eac..fe0b108203b 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer_routing.rs @@ -12,6 +12,7 @@ use azure_core_amqp::{message::AmqpMessageBody, AmqpList}; use azure_core_test::{recorded, TestContext}; use azure_messaging_eventhubs::{ + error::ErrorKind, models::{AmqpMessage, AmqpSimpleValue, AmqpValue, EventData, ReceivedEventData}, ConsumerClient, EventDataBatchOptions, EventReceiver, OpenReceiverOptions, ProducerClient, SendEventOptions, StartLocation, StartPosition, @@ -31,7 +32,6 @@ const ROUTED_EVENT_COUNT: usize = 5; // ProducerCanSendSingleLargeEventInASet uses new byte[100000] with the comment // "Actual limit is 1046520 for a single event". const LARGE_BODY_LEN: usize = 100_000; -const OVERSIZED_BODY_LEN: usize = 2 * 1024 * 1024; const SEQUENCE_INT_VALUE: i32 = 1_234_567_890; const BINARY_PROPERTY_KEY: &str = "producer-routing-binary"; const BINARY_PROPERTY_VALUE: [u8; 4] = [0x12, 0x34, 0x56, 0x78]; @@ -41,7 +41,6 @@ const BINARY_PROPERTY_VALUE: [u8; 4] = [0x12, 0x34, 0x56, 0x78]; // ordering. const PARTITION_SWEEP_DEADLINE: Duration = Duration::from_secs(20); const READ_DEADLINE: Duration = Duration::from_secs(60); -const OVERSIZED_SEND_DEADLINE: Duration = Duration::from_secs(45); /// Returns the label of an event when the event belongs to this run. /// @@ -524,58 +523,72 @@ async fn large_event_body_round_trips_intact(ctx: TestContext) -> Result<(), Box Ok(()) } -/// A single event above the service cap must not send. +/// A batch refuses an event above the negotiated link maximum. /// -/// The batch half of this behavior lives in the other file: -/// create_batch_honors_max_size_in_bytes and -/// create_batch_rejects_size_above_link_maximum in tests/eventhubs_producer.rs -/// cover the batch refusal. This test covers the send_event path only. +/// This mirrors the .NET `ProducerCannotSendSetLargerThanMaximumSize`, which +/// also asserts on a set rather than on one event. The maximum is read from the +/// `InvalidBatchSize` error rather than hardcoded, so the test holds on every +/// tier: Basic caps a publication at 256 KB, Standard and Premium at 1 MB, and +/// Dedicated at 20 MB. +/// +/// +/// `ProducerClient::send_event` does NOT enforce this maximum. A live run +/// against a Standard namespace whose link reported `max_allowed: 1048576` +/// accepted a 2 MiB single event, returned `Ok(())`, and moved the partition +/// tail. That divergence between the batch path and the single event path +/// needs its own issue and its own fix, so no test here pins it. #[recorded::test(live)] -async fn oversized_send_event_does_not_succeed(ctx: TestContext) -> Result<(), Box> { - const TEST_NAME: &str = "oversized_send_event_does_not_succeed"; +async fn batch_refuses_event_above_the_link_maximum( + ctx: TestContext, +) -> Result<(), Box> { + const TEST_NAME: &str = "batch_refuses_event_above_the_link_maximum"; const PARTITION: &str = "0"; + const ABSURD_BATCH_SIZE: u64 = 512 * 1024 * 1024; let producer = open_producer(&ctx, TEST_NAME).await?; - let run_marker = format!("{TEST_NAME}-{}", azure_core::Uuid::new_v4()); - let body = large_body(OVERSIZED_BODY_LEN); - - // The test asserts nothing about the error kind or the message text. The - // error module has no size limit variant, and nothing pins how the fe2o3 - // link payload error surfaces. Tighten this to the real error after a live - // run shows the kind. - let outcome = tokio::time::timeout( - OVERSIZED_SEND_DEADLINE, - producer.send_event( - EventData::builder() - .with_body(body) - .add_property(RUN_MARKER_KEY.to_string(), run_marker.clone()) - .build(), - Some(SendEventOptions { - partition_id: Some(PARTITION.to_string()), - }), - ), - ) - .await; - match outcome { - Err(_elapsed) => info!( - "run {run_marker}: the {OVERSIZED_BODY_LEN} byte send to partition {PARTITION} did not \ - finish within {OVERSIZED_SEND_DEADLINE:?}, so it did not succeed." - ), - Ok(Err(err)) => info!( - "run {run_marker}: the service rejected the {OVERSIZED_BODY_LEN} byte send to \ - partition {PARTITION}. {err:?}" - ), - Ok(Ok(())) => panic!( - "run {run_marker}: a {OVERSIZED_BODY_LEN} byte event reached partition {PARTITION} \ - within {OVERSIZED_SEND_DEADLINE:?}, but the Standard tier caps a single event at \ - about 1 MB" - ), - } + // Ask for a batch far above any link maximum. The error carries the real + // maximum, which is the only public way to read it. + let error = producer + .create_batch(Some(EventDataBatchOptions { + max_size_in_bytes: Some(ABSURD_BATCH_SIZE), + partition_id: Some(PARTITION.to_string()), + ..Default::default() + })) + .await + .err() + .expect("a batch above the link maximum must be refused"); + + let max_allowed = match error.kind { + ErrorKind::InvalidBatchSize { max_allowed, .. } => max_allowed, + other => panic!("expected InvalidBatchSize, got {other:?}"), + }; + info!("{TEST_NAME}: the link reports max_allowed {max_allowed} bytes."); + assert!( + max_allowed > 0, + "the link maximum must be positive, got {max_allowed}" + ); + + let batch = producer + .create_batch(Some(EventDataBatchOptions { + partition_id: Some(PARTITION.to_string()), + ..Default::default() + })) + .await?; - // The oversized send can already have detached the link, so a close error is - // not a test failure. - let _ = producer.close().await; + let oversized = usize::try_from(max_allowed).expect("link maximum must fit in a usize") + 1; + let added = batch.try_add_event_data( + EventData::builder() + .with_body(large_body(oversized)) + .build(), + None, + )?; + assert!( + !added, + "a {oversized} byte event must not fit a batch whose maximum is {max_allowed}" + ); + + producer.close().await?; Ok(()) }