hubcio commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4061555230
##########
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:
nit: this tag change breaks plugins built against 0.4.0, but the
breaking-change list at line 57 still says two items. add it there with the
rebuild rule.
##########
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:
nit: the README behavior paragraph (line 59) still says proto is skipped as
unsupported. say that proto text holding JSON indexes as a document and other
proto text lands in `text`.
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -157,6 +158,72 @@ pub enum Payload {
}
impl Payload {
+ /// The `Schema` describing this payload's variant.
Review Comment:
nit: the run contract - several `consume()` calls per poll, each repeating
the same `current_offset` - lives only in the sinks README. put it on
`Sink::consume` (line 138) and `MessagesMetadata` (line 376), where plugin
authors read.
##########
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:
nit: `route_value` parses the proto text and drops the document, then
`write_data_files` parses the same text again at router/mod.rs:156. normalize
the payload to `Payload::Json` once at the top of the `route_data` loop.
##########
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:
simplification: this is the same four-payload transform `split_batch()`
builds at line 1068. call the helper like the other two tests do.
##########
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:
simplification: this four-line normalizer is copied into five sinks. add one
consuming helper on `Payload` next to `json_document` and call it at each site.
also at elasticsearch_sink/src/lib.rs:322, http_sink/src/lib.rs:386,
quickwit_sink/src/lib.rs:320, surrealdb_sink/src/lib.rs:901.
--
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]
