mlevkov commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4043706088
##########
core/connectors/sinks/clickhouse_sink/src/body.rs:
##########
@@ -98,7 +98,10 @@ pub(crate) fn build_string_body(
let mut buf = Vec::with_capacity(messages.len() * 64);
for msg in messages {
match &msg.payload {
- Payload::Text(s) => {
+ // `Payload::Proto` holds proto text, so a passthrough body takes
+ // it the same way it takes `Payload::Text`. The JSON and RowBinary
+ // builders above cannot: they need a parsed document.
Review Comment:
This comment is the stated reason the JSON and RowBinary builders still skip
`Payload::Proto`, and it is wrong for the JSON one. `build_json_body` writes a
serialized document and a newline (`body.rs:46`), and on this path the proto
text is already a serialized JSON document (`proto_convert.rs:630`). The
RowBinary half is correct, because `binary.rs:52` needs a parsed object.
The result is silent loss. `JsonEachRow` is `#[default]`
(`clickhouse_sink/src/lib.rs:60`), the skip still returns `Ok` (`sink.rs:146`),
and the runtime adds the whole run to `iggy_connector_messages_processed_total`
while `iggy_connector_errors_total` stays flat.
`AutoCommitWhen::PollingMessages` (`runtime/src/sink.rs:515`) commits the
offset before the plugin sees the batch, so nothing replays. On master these
rows loaded, because the `Schema::Json` tag reparsed the text. RowBinary also
loses a signal it had: that path failed the batch and counted an error before,
and now drops the row quietly.
This does not need a missing descriptor. `encode_json_with_schema` returns
`Err` for any non-object top-level JSON (`proto_convert.rs:285`), so a fully
configured pipeline takes the same fallback on every array, string, number,
boolean and null message.
Fix: parse `Payload::Proto(text)` as JSON in both builders, then take the
existing paths. Parse and re-serialize rather than pass the text through,
because `pretty_json` (`proto_convert.rs:48`) puts newlines in it and
JSONEachRow needs one object per line.
##########
core/connectors/sinks/doris_sink/src/lib.rs:
##########
@@ -1098,8 +1098,9 @@ impl Sink for DorisSink {
//
// The lone hard-abort is a non-JSON payload (via `?`): a stream-wide
// schema-contract violation, not a transient chunk failure. Under the
- // documented `schema = "json"` config the SDK drops non-JSON before
- // consume() is called, so this stands as a defensive guard.
+ // documented `schema = "json"` config the runtime's JSON decoder drops
+ // non-JSON bytes before consume() is called, so this stands as a
+ // defensive guard unless a format-converting transform is configured.
Review Comment:
This edit names the hole and leaves it open. `proto_convert` is a
format-converting transform, and with no descriptor it returns `Payload::Proto`
holding a JSON document (`proto_convert.rs:656`).
Doris then takes the `_` arm at `:1109`, and the `?` at `:1117` ends the
whole `consume()` call, so no chunk is written. The offset already committed
(`runtime/src/sink.rs:515`), and the runtime has no replay and no dead-letter
queue, so the batch is gone and the loss repeats every poll. On master the same
configuration loaded rows, because the `Schema::Json` tag reparsed the text.
The config is not hypothetical: this PR adds one at
`proto_text_config/elasticsearch_sink.toml:41`.
Delta (`delta_sink/src/sink.rs:110`) and Iceberg
(`iceberg_sink/src/router/mod.rs:152`) have the same shape. Elasticsearch,
Meilisearch and the ClickHouse passthrough already take this payload, so three
sinks accept it and three fail.
Fix: parse `Payload::Proto` as JSON in Doris, Delta and Iceberg, and keep
the abort when the parse fails. Do not treat all proto text as JSON:
`proto_convert.rs:695` and `:714` produce genuine proto text that is not JSON.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -674,6 +680,10 @@ async fn process_messages(
continue;
};
+ // Read the tag off the payload before `try_into_vec` consumes it. The
+ // decoder's own schema names the format it reads, not the variant it
+ // returned, and a transform may have changed the variant since.
+ let schema = message.payload.schema();
Review Comment:
Three of the six stream `schema` values change the tag here, and the
Compatibility section names one. Under their defaults, `avro` (`avro.rs:40`),
`flat_buffer` (`flatbuffer.rs:42`) and `proto` (`proto.rs:48`) all decode to
`Payload::Json`, so all three change. `json`, `text` and `raw` do not, which
the section states correctly.
For `flat_buffer` this is the same output change in the same three sinks the
section already describes for avro: S3, HTTP and SurrealDB stop base64-encoding
and write the document. For `proto` the effect differs, and one case improves,
because a `proto` stream failed every poll into Doris before and now writes
rows.
Fix: add `flat_buffer` and `proto` to the Compatibility section.
##########
core/connectors/sinks/elasticsearch_sink/src/lib.rs:
##########
@@ -381,7 +381,9 @@ impl Sink for ElasticsearchSink {
}
}
}
- Payload::Text(text) => simd_json::json!({
+ // `Payload::Proto` holds proto text, so it indexes the
+ // same way `Payload::Text` does.
+ Payload::Text(text) | Payload::Proto(text) =>
simd_json::json!({
Review Comment:
This arm indexes proto text as one opaque `text` field. On a `schema =
"json"` stream with a descriptor-less `proto_convert`, the old tag reparsed the
same bytes and Elasticsearch indexed a structured document, so for that
pipeline this is a searchability regression rather than a new capability.
Fix: try a JSON parse on `Payload::Proto` first, and fall back to this arm
when it fails.
##########
core/connectors/sdk/README.md:
##########
@@ -175,7 +177,7 @@ These are SDK configuration fields, not Random or Stdout
`plugin_config` keys. T
#### Transform Options
- **`proto_convert`**: Transform for converting between protobuf and other
formats
-- **`source_format`** / **`target_format`**: Formats to convert between - any
schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`)
+- **`source_format`** / **`target_format`**: Formats to convert between - any
schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`).
`source_format` must match the variant the decoder or a preceding transform
actually produced, or the message is rejected
Review Comment:
Two problems in this sentence.
`proto_convert` does not reject a mismatched `source_format`.
`ProtoConvert::transform` (`proto_convert.rs:737`) dispatches on the format
pair with no up-front guard, and the example at `README.md:128` depends on
that: `schema = "raw"` with `source_format = "proto"` reaches
`protobuf_to_json`, whose `Payload::Raw` arm always returns `Ok`.
`flatbuffer_convert.rs:114` is the only transform that has the guard, and
`avro_convert` has none either.
"A preceding transform" also implies an order the runtime does not
guarantee. `runtime/src/transform.rs:26` builds the chain from a `HashMap`, so
the order varies per process.
Fix: limit the claim to `flatbuffer_convert`, and either drop the ordering
wording or state that the chain order is not defined.
##########
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:
Nothing records that a batch split. `stage_ffi` (`:392`) records total
duration, and the per-run log at `:797` fires only on failure, so an operator
cannot see one poll become N calls. A batch that alternates variants makes one
FFI call and one sink round trip per message.
Fix: record the run count per batch as a metric. A log line on every split
gets noisy once a descriptor-backed `proto_convert` splits most batches, so the
metric is the useful half.
##########
core/connectors/sinks/README.md:
##########
@@ -249,6 +249,8 @@ for message in messages {
While the schema of messages (that will be consumed from the Iggy stream),
cannot be controlled by the sink connector itself, the built-in configuration
allows to decide what's the expected format of the messages (the particular
`StreamDecoder` will be used).
+`messages_metadata.schema` names the variant each `Payload` in the batch
actually holds, which is not always the stream's `schema` setting. A decoder
may return a different form than the wire format it reads: with `schema =
"avro"` the Avro decoder extracts to JSON by default, so the batch arrives as
`Payload::Json`. A configured transform can change the variant again. Match on
the `Payload` itself and treat `messages_metadata.schema` as a description of
what arrived, not of how the stream was configured.
Review Comment:
One poll can now reach a plugin as several `consume()` calls, and this
paragraph does not say so. `runtime/src/sink.rs:761` loops over the runs and
`:764` gives every run the same `current_offset`, so a sink that commits per
call applies part of a poll. Delta commits a table version per call, and
Iceberg commits per call.
Nothing in the repository breaks today, because no sink keys its output on
`current_offset`. It is still a new contract for plugin authors.
One exception is worth a sentence: a fully dropped batch is tagged with the
decoder wire format (`runtime/src/sink.rs:748`), so an avro stream's tag flips
between `avro` and `json`.
Fix: state that a mixed batch arrives as several `consume()` calls that
repeat one offset, and note the empty-batch case.
##########
core/connectors/sdk/src/sink.rs:
##########
@@ -196,7 +198,12 @@ impl<T: Sink + std::fmt::Debug> SinkContainer<T> {
}
};
- let payload = match
messages_metadata.schema.try_into_payload(message.payload) {
+ // The runtime tags each run from `Payload::schema`, so the
+ // tag names a variant here rather than a wire format.
+ let payload = match Payload::try_from_schema(
Review Comment:
Two things to record against this line.
The meaning of `Schema::Proto` in `MessagesMetadata` changed and
`iggy_connector_sdk` stays at 0.4.0, so a plugin built before this change still
calls `Schema::try_into_payload` and rebuilds a Proto run as `Payload::Raw`. A
version bump alone does not repair that, because a rebuilt Doris, Delta or
ClickHouse sink still rejects the `Payload::Proto` it now receives. A runtime
check is not available either: `iggy_sink_version` (`sdk/src/sink.rs:328`)
expands `CARGO_PKG_VERSION` inside the plugin crate, so it reports the plugin
version, and the runtime only stores the string.
This line also adds a per-message JSON parse on avro and flatbuffer streams.
The old `Schema::Avro` arm moved the bytes without reading them, and the new
`Schema::Json` arm runs `simd_json::to_owned_value` per message. That is the
price of the fix and not a defect, but operators sizing sink CPU need to know.
Fix: bump to 0.5.0, and document both the rebuild requirement and the new
parse.
##########
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:
This round-trip test compares only the rebuilt variant, so a
`try_from_schema` that drops or corrupts the value still passes. The test
proves the tag survives, not the payload.
Fix: compare the rebuilt payload bytes against the original bytes.
##########
core/connectors/sinks/meilisearch_sink/src/lib.rs:
##########
@@ -360,7 +360,9 @@ impl MeilisearchSink {
]),
}
}
- Payload::Text(text) => Map::from_iter([
+ // `Payload::Proto` holds proto text, so it indexes the same
+ // way `Payload::Text` does.
+ Payload::Text(text) | Payload::Proto(text) => Map::from_iter([
Review Comment:
Same shape as the Elasticsearch arm: proto text is stored under a single
`text` key, so the document loses every original field. The same
descriptor-less `proto_convert` pipeline stored a structured document before.
Fix: try a JSON parse on `Payload::Proto` first, and fall back to this arm
when it fails.
##########
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:
The new per-run subtraction has no coverage. No test in this module fails a
run, so `processed_count -= run_len` never executes, and a wrong sign or a
wrong operand would not fail the suite.
Fix: add a stub sink that fails one run of a split batch, and assert the
resulting `processed_count`.
##########
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:
This assertion locks in the flattened result as the expected one. It asserts
`data_type == "text"`, which is the shape the document takes only after the
structured fields are gone, so the test passes precisely when the regression
happens.
Fix: assert that the indexed document keeps the original message fields.
--
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]
