MarcusKainth commented on code in PR #4204: URL: https://github.com/apache/iggy/pull/4204#discussion_r4058295206
########## core/integration/tests/connectors/elasticsearch/proto_text.rs: ########## @@ -0,0 +1,110 @@ +// 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 super::TEST_MESSAGE_COUNT; +use crate::connectors::create_test_messages; +use crate::connectors::fixtures::ElasticsearchSinkFixture; +use bytes::Bytes; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::Identifier; +use iggy_common::MessageClient; +use integration::harness::seeds; +use integration::iggy_harness; + +/// A `proto_convert` transform with no descriptor falls back to proto text, so +/// the batch is tagged `Schema::Proto` and the sink is handed `Payload::Proto`. +/// Before the sink SDK gained its own inverse of `Payload::schema`, that tag +/// rebuilt as `Payload::Raw`, and before Elasticsearch took Proto as text it +/// dropped the batch on its catch-all arm. +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/elasticsearch/proto_text.toml")), + seed = seeds::connector_stream +)] +async fn given_a_proto_convert_transform_when_the_sink_consumes_should_index_the_payload( + harness: &TestHarness, + fixture: ElasticsearchSinkFixture, +) { + let client = harness.root_client().await.unwrap(); + + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let messages_data = create_test_messages(TEST_MESSAGE_COUNT); + let mut messages: Vec<IggyMessage> = messages_data + .iter() + .enumerate() + .map(|(i, msg)| { + let payload = serde_json::to_vec(msg).expect("Failed to serialize message"); + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_documents(TEST_MESSAGE_COUNT) + .await + .expect("the proto text batch must be indexed, not dropped"); + + fixture + .refresh_index() + .await + .expect("Failed to refresh index"); + + let search_result = fixture + .search_documents() + .await + .expect("Failed to search documents"); + + assert_eq!( + search_result.hits.total.value, TEST_MESSAGE_COUNT, + "Expected {TEST_MESSAGE_COUNT} documents in Elasticsearch" + ); + + for hit in &search_result.hits.hits { + assert_eq!( + hit.source.get("data_type").and_then(|v| v.as_str()), Review Comment: Yes, it asserted the regression. I have rewritten it to assert the original fields are indexed and no `data_type` wrapper exists. The useful property: it passes on `master`, fails on `d8a8ece`, and passes again after `30216eb`. ########## core/connectors/runtime/src/sink.rs: ########## @@ -721,43 +757,50 @@ 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 { Review Comment: Added two tests through a stub that fails whichever run's schema the test registers. One failing run in a split batch debits only that run and the later runs still execute, and every run failing reports zero processed with one error per run. The subtraction is gone as well; `processed_count` now accumulates in the success branch, so the underflow class cannot come back. `2dbebc4`. ########## core/connectors/sdk/src/lib.rs: ########## @@ -471,3 +522,103 @@ pub enum Error { #[error("State provider latched after a permanent state error")] StateLatched, } + +#[cfg(test)] +mod tests { + use super::*; + + fn all_payloads() -> Vec<(Payload, Schema)> { + vec![ + (Payload::Json(simd_json::json!({"id": 1})), Schema::Json), + (Payload::Json(simd_json::json!([1, 2, 3])), Schema::Json), + (Payload::Json(simd_json::json!("scalar")), Schema::Json), + (Payload::Json(simd_json::json!(null)), Schema::Json), + (Payload::Raw(vec![1, 2, 3]), Schema::Raw), + (Payload::Raw(Vec::new()), Schema::Raw), + (Payload::Text("hello".to_owned()), Schema::Text), + (Payload::Text(String::new()), Schema::Text), + (Payload::Proto("proto text".to_owned()), Schema::Proto), + (Payload::Proto(String::new()), Schema::Proto), + (Payload::FlatBuffer(vec![4, 5, 6]), Schema::FlatBuffer), + (Payload::FlatBuffer(Vec::new()), Schema::FlatBuffer), + (Payload::Avro(vec![7, 8, 9]), Schema::Avro), + (Payload::Avro(Vec::new()), Schema::Avro), + ] + } + + #[test] + fn given_every_payload_variant_when_schema_is_read_should_name_that_variant() { + for (payload, expected) in all_payloads() { + assert_eq!( + payload.schema(), + expected, + "wrong schema for {payload}, expected {expected}" + ); + } + } + + #[test] + fn given_a_payload_when_round_tripped_through_its_own_schema_should_keep_the_variant() { + for (payload, schema) in all_payloads() { + let bytes = payload + .try_into_vec() + .unwrap_or_else(|error| panic!("failed to serialize {schema} payload: {error}")); + let rebuilt = Payload::try_from_schema(schema, bytes) + .unwrap_or_else(|error| panic!("failed to rebuild {schema} payload: {error}")); + + assert_eq!( Review Comment: Correct, it proved the tag and not the payload. It now compares `try_to_bytes()` of the rebuilt payload against the original for all fourteen rows (`30216eb`). I have kept the two inverses separate rather than delegating five arms, because the docs frame them as inverses of two different tags and the byte-level test is now the drift guard. The `Any` test also asserts `type_url` and the base64 `value` rather than key presence. `d94bedd`. ########## core/connectors/runtime/src/sink.rs: ########## @@ -721,43 +757,50 @@ 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 { Review Comment: Added `iggy_connector_sink_runs`, a counter incremented by the run count of each batch. `iggy_connector_stage_duration_seconds_count{stage="total"}` already counts batches, so runs per batch is the ratio of the two, and a histogram would have cost eleven series per sink key for the same answer. The `ffi` stage sample stays one per batch with the summed duration, which is what you described and is identical to `master` for a uniform batch; the README now says so next to the counter. The benchmark event carries a `runs` field too. `2dbebc4`. -- 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]
