MarcusKainth commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4069986087
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -157,6 +158,72 @@ pub enum Payload {
}
impl Payload {
+ /// The `Schema` describing this payload's variant.
Review Comment:
`Sink::consume` `L138` now states the run split, the repeated
`current_offset`, and the empty batch that still arrives as one call carrying
no messages. `MessagesMetadata` `L392` documents all three fields, with
`current_offset` as the partition's high-water offset at the time of the poll
and `schema` as the variant the payloads hold rather than the stream's
configured format. `sinks/README.md` `L252-258` keeps the long form. A new
runtime test pins the empty-batch call, since the rustdoc now promises it.
`093f5eb`.
##########
core/connectors/sdk/README.md:
##########
@@ -156,12 +156,16 @@ Create the stream and topic, then start the runtime from
the repository root:
IGGY_CONNECTORS_CONFIG_PATH=connectors.toml ./target/release/iggy-connectors
```
-The source sends 100 records, then continues polling without new messages.
Stdout logs message offsets and the serialized JSON envelope bytes, containing
`type_url` and base64 `value`. The sink's `raw` schema also determines how the
plugin receives those transformed bytes.
+The source sends 100 records, then continues polling without new messages.
Stdout logs message offsets and the serialized JSON envelope bytes, containing
`type_url` and base64 `value`. The batch handed to the plugin is tagged with
the payload's own schema rather than the stream's configured `raw`:
`proto_convert` leaves a `Payload::Json`, so the plugin receives a `json` batch.
Review Comment:
The lead-in at `sdk/README.md` `L57` now reads as two removals and one field
changing meaning, and a paragraph at `L66` records that
`MessagesMetadata.schema` and `RawMessages.schema` name the variant the
`Payload` holds rather than the wire format the stream's decoder reads.
`Schema::Proto` means a `Payload::Proto` string on the sink path where it meant
protobuf wire bytes before, so a plugin built against 0.4.0 rebuilds the run as
the wrong variant with no error anywhere. The rebuild rule stays where it was,
in the lead-in. `093f5eb`.
##########
core/connectors/sinks/meilisearch_sink/src/lib.rs:
##########
@@ -360,7 +367,7 @@ impl MeilisearchSink {
]),
}
}
- Payload::Text(text) => Map::from_iter([
+ Payload::Text(text) | Payload::Proto(text) => Map::from_iter([
Review Comment:
The `## Behavior` paragraph now says proto text holding a JSON document is
indexed as that document, and proto text that is not JSON is indexed in the
`text` field alongside text payloads. That leaves Avro and FlatBuffer as the
schemas skipped with a warning, which matches `lib.rs` `L368-373`. `093f5eb`.
##########
core/connectors/sinks/meilisearch_sink/src/lib.rs:
##########
@@ -339,6 +339,13 @@ impl MeilisearchSink {
payload,
} = message;
+ // The descriptor-less `proto_convert` fallback puts a JSON document in
+ // `Payload::Proto`. It is indexed as that document, and the text arm
+ // below is kept for proto text that is not JSON.
+ let payload = match payload.json_document() {
Review Comment:
`Payload::into_json_document` sits next to `json_document` in
`sdk/src/lib.rs` `L241`, and the five sites call it: Meilisearch `L342`,
Elasticsearch `L317`, HTTP `L381`, Quickwit `L316`, SurrealDB `L896`. Behaviour
is unchanged at every site and the existing per-sink tests cover it. Four SDK
unit tests pin proto text holding JSON, proto text that is not JSON with the
text left byte-identical, a JSON payload returned as it was, and every other
variant returned with both its schema and its bytes intact. `535214c`.
##########
core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs:
##########
@@ -83,19 +83,25 @@ impl DynamicRouter {
}
fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String>
{
- match &message.payload {
- Payload::Json(payload) => payload
- .as_object()
- .and_then(|obj| obj.get(&self.route_field))
- .map(|val| val.to_string()),
- _ => {
- warn!("Unsupported format for iceberg connector");
- None
- }
- }
+ route_value(&message.payload, &self.route_field)
}
}
+/// The routing value a payload carries under `route_field`, read from the JSON
+/// document it holds. Proto text holding JSON is the descriptor-less
+/// `proto_convert` fallback and routes like the document it holds; a payload
+/// with no document does not route.
+fn route_value(payload: &Payload, route_field: &str) -> Option<String> {
+ let Some(document) = payload.json_document() else {
Review Comment:
`route_data` normalises each message at the top of the loop, so
`route_value` and `write_data_files` both read the document off a
`Payload::Json`, and proto text holding JSON is parsed once on the dynamic path
rather than twice. `route_value`'s doc at `L90-93` is updated to match. The
`json_document` call at `router/mod.rs` `L156` serves
`StaticRouter::route_data`, which reaches `write_data_files` with no routing
parse and so parses once either way. The three `route_value` tests at
`L183-202`, `router/mod.rs` `L448`, and the `connectors::iceberg` suite pass
unchanged. `535214c`.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -721,53 +761,556 @@ async fn process_messages(
RuntimeError::FailedToSerializeTopicMetadata
})?;
- let messages_meta =
postcard::to_allocvec(&messages_metadata).map_err(|error| {
- error!(
- "Failed to serialize messages metadata for sink connector with ID:
{plugin_id}. {error}"
- );
- RuntimeError::FailedToSerializeMessagesMetadata
- })?;
+ let mut ffi_elapsed = Duration::ZERO;
+ for (schema, run) in runs {
+ let messages_metadata = MessagesMetadata {
+ partition_id,
+ current_offset,
+ schema,
+ };
+ let messages_meta =
postcard::to_allocvec(&messages_metadata).map_err(|error| {
+ error!(
+ "Failed to serialize messages metadata for sink connector with
ID: {plugin_id}. {error}"
+ );
+ RuntimeError::FailedToSerializeMessagesMetadata
+ })?;
- let messages = postcard::to_allocvec(&RawMessages {
- schema: decoder.schema(),
- messages,
- })
- .map_err(|error| {
- error!("Failed to serialize messages for sink connector with ID:
{plugin_id}. {error}");
- RuntimeError::FailedToSerializeRawMessages
- })?;
+ let run_len = run.len();
+ let messages = postcard::to_allocvec(&RawMessages {
+ schema,
+ messages: run,
+ })
+ .map_err(|error| {
+ error!("Failed to serialize messages for sink connector with ID:
{plugin_id}. {error}");
+ RuntimeError::FailedToSerializeRawMessages
+ })?;
- let ffi_start = Instant::now();
- let result = (consume)(
- plugin_id,
- topic_meta.as_ptr(),
- topic_meta.len(),
- messages_meta.as_ptr(),
- messages_meta.len(),
- messages.as_ptr(),
- messages.len(),
- );
- let ffi_elapsed = ffi_start.elapsed();
- let processed_count = if result == 0 {
- processed_count
- } else {
- error!(
- "Failed to consume {processed_count} messages for sink connector
with ID: {plugin_id}, stream: {}, topic: {}, status: {result}",
- topic_metadata.stream, topic_metadata.topic
+ let ffi_start = Instant::now();
+ let result = (consume)(
+ plugin_id,
+ topic_meta.as_ptr(),
+ topic_meta.len(),
+ messages_meta.as_ptr(),
+ messages_meta.len(),
+ messages.as_ptr(),
+ messages.len(),
);
- metrics.inc_errors_with_labels(&labels.counter);
- 0
- };
+ ffi_elapsed += ffi_start.elapsed();
+ if result == 0 {
+ processed_count += run_len;
+ } else {
+ error!(
+ "Failed to consume {run_len} messages for sink connector with
ID: {plugin_id}, stream: {}, topic: {}, schema: {schema}, status: {result}",
+ topic_metadata.stream, topic_metadata.topic
+ );
+ metrics.inc_errors_with_labels(&labels.counter);
+ }
+ }
+
+ // Counted once every call has been made, so a serialisation failure above
+ // cannot leave calls on the counter that never happened.
+ metrics.inc_sink_runs_with_labels(&labels.counter, run_count as u64);
Ok(SinkBatchTiming {
processed_count,
+ runs: run_count,
decode_elapsed,
ffi_elapsed,
})
}
struct SinkBatchTiming {
processed_count: usize,
+ /// FFI calls the batch was split into, one per contiguous payload variant.
+ runs: usize,
decode_elapsed: Duration,
+ /// Summed over every run, so one batch is one `stage_ffi` sample however
+ /// many calls it took.
ffi_elapsed: Duration,
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use dashmap::DashMap;
+ use iggy::prelude::IggyMessageHeader;
+ use iggy_connector_sdk::transforms::TransformType;
+ use iggy_connector_sdk::transforms::{ProtoConvert, ProtoConvertConfig};
+ use iggy_connector_sdk::{Error, Payload};
+ use prost::Message as _;
+ use std::sync::LazyLock;
+ use std::sync::atomic::AtomicU32;
+
+ /// One entry per FFI call the stub sink received, keyed by plugin id so
+ /// tests sharing the binary do not read each other's batches.
+ static CONSUMED: LazyLock<DashMap<u32, Vec<ConsumedBatch>>> =
LazyLock::new(DashMap::new);
+
+ static TEST_PLUGIN_ID: AtomicU32 = AtomicU32::new(u32::MAX / 2);
+
+ struct ConsumedBatch {
+ metadata_schema: Schema,
+ messages_schema: Schema,
+ payloads: Vec<Vec<u8>>,
+ offsets: Vec<u64>,
+ }
+
+ /// Runs the stub sink fails, keyed by plugin id and matched on the run's
+ /// schema, so a test reads the failure off the run split rather than off
+ /// a call counter.
+ static FAILING_SCHEMAS: LazyLock<DashMap<u32, Schema>> =
LazyLock::new(DashMap::new);
+
+ /// High-water offset deliberately unrelated to any message offset in a
+ /// batch: `current_offset` is the partition's head from the poll, and a
+ /// helper that made it equal the last message's offset would let a test
+ /// assert the wrong contract without failing.
+ const TEST_CURRENT_OFFSET: u64 = 10_000;
+
+ /// Records the batch the FFI call carried and returns its run schema.
+ fn capture_batch(
+ plugin_id: u32,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> Schema {
+ let messages_meta =
+ unsafe { std::slice::from_raw_parts(messages_meta_ptr,
messages_meta_len) };
+ let messages = unsafe { std::slice::from_raw_parts(messages_ptr,
messages_len) };
+ let metadata = postcard::from_bytes::<MessagesMetadata>(messages_meta)
+ .expect("failed to deserialize messages metadata");
+ let raw = postcard::from_bytes::<RawMessages>(messages).expect("failed
to deserialize");
+
+ CONSUMED.entry(plugin_id).or_default().push(ConsumedBatch {
+ metadata_schema: metadata.schema,
+ messages_schema: raw.schema,
+ offsets: raw.messages.iter().map(|message|
message.offset).collect(),
+ payloads: raw
+ .messages
+ .into_iter()
+ .map(|message| message.payload)
+ .collect(),
+ });
+ metadata.schema
+ }
+
+ extern "C" fn capturing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ 0
+ }
+
+ /// Captures like `capturing_consume`, then fails the run whose schema the
+ /// test registered in `FAILING_SCHEMAS`.
+ extern "C" fn selectively_failing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ let schema = capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ match FAILING_SCHEMAS.get(&plugin_id) {
+ Some(failing) if *failing == schema => 1,
+ _ => 0,
+ }
+ }
+
+ extern "C" fn always_failing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ 1
+ }
+
+ /// Rewrites every payload to the configured variant, standing in for a
+ /// third-party transform that changes the payload type.
+ struct RetaggingTransform {
+ payloads: Vec<Payload>,
+ next: std::sync::Mutex<usize>,
+ }
+
+ impl Transform for RetaggingTransform {
+ fn r#type(&self) -> TransformType {
+ TransformType::AvroConvert
+ }
+
+ fn transform(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ let mut next = self.next.lock().expect("transform counter was
poisoned");
+ message.payload = self.payloads[*next %
self.payloads.len()].clone();
+ *next += 1;
+ Ok(Some(message))
+ }
+ }
+
+ fn next_plugin_id() -> u32 {
+ TEST_PLUGIN_ID.fetch_add(1, Ordering::Relaxed)
+ }
+
+ fn test_message(offset: u64, payload: Vec<u8>) -> IggyMessage {
+ IggyMessage {
+ header: IggyMessageHeader {
+ checksum: 0,
+ id: u128::from(offset) + 1,
+ offset,
+ timestamp: 0,
+ origin_timestamp: 0,
+ user_headers_length: 0,
+ payload_length: payload.len() as u32,
+ reserved: 0,
+ },
+ payload: payload.into(),
+ user_headers: None,
+ }
+ }
+
+ fn avro_schema() -> apache_avro::Schema {
+ apache_avro::Schema::parse_str(
+
r#"{"type":"record","name":"Event","fields":[{"name":"id","type":"long"}]}"#,
+ )
+ .expect("failed to parse Avro schema")
+ }
+
+ fn avro_datum(schema: &apache_avro::Schema, id: i64) -> Vec<u8> {
+ let record = apache_avro::types::Value::Record(vec![(
+ "id".to_owned(),
+ apache_avro::types::Value::Long(id),
+ )]);
+ apache_avro::writer::datum::GenericDatumWriter::builder(schema)
+ .build()
+ .expect("failed to build Avro writer")
+ .write_value_to_vec(record)
+ .expect("failed to encode Avro datum")
+ }
+
+ async fn run(
+ plugin_id: u32,
+ decoder: Arc<dyn StreamDecoder>,
+ transforms: Vec<Arc<dyn Transform>>,
+ messages: Vec<IggyMessage>,
+ ) -> SinkBatchTiming {
+ run_with(plugin_id, capturing_consume, decoder, transforms, messages)
+ .await
+ .0
+ }
+
+ /// Drives one batch through `process_messages` with the given stub sink
+ /// and hands back the metrics it wrote, so a test can read counters.
+ async fn run_with(
+ plugin_id: u32,
+ consume: ConsumeCallback,
+ decoder: Arc<dyn StreamDecoder>,
+ transforms: Vec<Arc<dyn Transform>>,
+ messages: Vec<IggyMessage>,
+ ) -> (SinkBatchTiming, Arc<Metrics>) {
+ let metrics = Arc::new(Metrics::init());
+ let labels = SinkLabels::new("test_sink");
+ let topic_metadata = TopicMetadata {
+ stream: "test_stream".to_owned(),
+ topic: "test_topic".to_owned(),
+ };
+
+ let timing = process_messages(
+ plugin_id,
+ 0,
+ TEST_CURRENT_OFFSET,
+ &topic_metadata,
+ messages,
+ &consume,
+ &transforms,
+ &decoder,
+ &metrics,
+ &labels,
+ )
+ .await
+ .expect("processing the batch should succeed");
+ (timing, metrics)
+ }
+
+ /// Four messages retagged `Text, Text, Raw, Text`, which the runtime
+ /// splits into three runs.
+ fn split_batch() -> (Arc<RetaggingTransform>, Vec<IggyMessage>) {
+ let transform = Arc::new(RetaggingTransform {
+ payloads: vec![
+ Payload::Text("first".to_owned()),
+ Payload::Text("second".to_owned()),
+ Payload::Raw(vec![9]),
+ Payload::Text("fourth".to_owned()),
+ ],
+ next: std::sync::Mutex::new(0),
+ });
+ let messages = (0..4)
+ .map(|offset| test_message(offset, br#"{"id":1}"#.to_vec()))
+ .collect();
+ (transform, messages)
+ }
+
+ fn captured(plugin_id: u32) -> Vec<ConsumedBatch> {
+ CONSUMED
+ .remove(&plugin_id)
+ .map(|(_, batches)| batches)
+ .unwrap_or_default()
+ }
+
+ #[tokio::test]
+ async fn
given_an_avro_stream_when_batch_is_tagged_should_use_the_decoded_payload_schema()
{
+ let plugin_id = next_plugin_id();
+ let schema = avro_schema();
+ let decoder = Arc::new(
+ AvroStreamDecoder::try_new(AvroConfig {
+ schema_json: Some(schema.canonical_form()),
+ ..AvroConfig::default()
+ })
+ .expect("failed to build the Avro decoder"),
+ );
+ let messages = vec![
+ test_message(0, avro_datum(&schema, 1)),
+ test_message(1, avro_datum(&schema, 2)),
+ ];
+
+ let timing = run(plugin_id, decoder, Vec::new(), messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 2);
+ assert_eq!(batches.len(), 1, "a uniform batch is one FFI call");
+ // The decoder reads Avro but returns JSON, so JSON is what the sink is
+ // told it has.
+ assert_eq!(batches[0].metadata_schema, Schema::Json);
+ assert_eq!(batches[0].messages_schema, Schema::Json);
+ for payload in &batches[0].payloads {
+
serde_json::from_slice::<serde_json::Value>(payload).expect("payload should be
JSON");
+ }
+ }
+
+ #[tokio::test]
+ async fn
given_a_transform_changing_the_variant_when_batch_is_tagged_should_follow_the_transform()
+ {
+ let plugin_id = next_plugin_id();
+ let transform = Arc::new(RetaggingTransform {
+ payloads: vec![Payload::Avro(vec![1, 2, 3])],
+ next: std::sync::Mutex::new(0),
+ });
+ let messages = vec![test_message(0, br#"{"id":1}"#.to_vec())];
+
+ run(plugin_id, Schema::Json.decoder(), vec![transform],
messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].metadata_schema, Schema::Avro);
+ assert_eq!(batches[0].messages_schema, Schema::Avro);
+ }
+
+ #[tokio::test]
+ async fn
given_mixed_payload_variants_when_batch_is_processed_should_split_into_runs() {
+ let plugin_id = next_plugin_id();
+ let transform = Arc::new(RetaggingTransform {
Review Comment:
The test calls `split_batch()`. `e587e4d`.
--
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]
