fix(connectors): tag sink batches with the payload's schema - #4204
MarcusKainth wants to merge 4 commits into
Conversation
A sink receives the wrong Payload variant when its stream is configured avro, proto or flatbuffer. The runtime tagged each batch with the decoder's schema, which names the wire format a decoder reads rather than the variant it returns, and the SDK rebuilds the payload from that tag alone. JSON-only sinks dropped the batch after the offset had been committed, and sinks that take the raw variants stored base64 of the JSON as Avro bytes. Read the tag from the payload the sink will actually receive, after transforms run. One Schema covers a whole FFI call, so messages are grouped into contiguous runs of the same variant and each run is sent on its own. Also corrects the sink documentation on where a payload's variant comes from, and the sdk protobuf example, which could not load. Closes apache#4053
Every new run reserved the messages still to come, so a batch that alternated payload variants reserved O(n^2) message slots. Only the first run is sized to the batch now; a later run is rare enough to grow on demand. Also pins the empty-batch fallback against a non-default schema, since asserting Json could not tell the stream's configured schema apart from Schema::default(), and waits for the payload log lines rather than the batch header in the integration test, which could otherwise read the log file between the two.
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #4204 +/- ##
=============================================
- Coverage 87.49% 68.66% -18.84%
Complexity 1575 1575
=============================================
Files 1280 1278 -2
Lines 223241 184062 -39179
Branches 186606 147427 -39179
=============================================
- Hits 195335 126387 -68948
- Misses 23196 52878 +29682
- Partials 4710 4797 +87
🚀 New features to boost your workflow:
|
| Payload::Json(_) => Schema::Json, | ||
| Payload::Raw(_) => Schema::Raw, | ||
| Payload::Text(_) => Schema::Text, | ||
| Payload::Proto(_) => Schema::Proto, |
There was a problem hiding this comment.
Payload::Proto contains text, but tagging it as Schema::Proto does not round-trip through the sink SDK: Schema::Proto::try_into_payload() reconstructs it as Raw (or Json for a protobuf Any), never Proto. The existing JSON→Proto transform can produce this text, so this change can cause a sink to receive Raw where it previously received Json and silently skip the message. Could we preserve the intended payload type across the FFI boundary and add a regression test for that transform path?
There was a problem hiding this comment.
Correct, and it is a regression against master, not just a round-trip gap: on a schema = "json" stream with a proto_convert transform the sink got Payload::Json on master and Payload::Raw on 31c0773. ClickHouse skips those rows; Delta, Doris and Iceberg fail the batch.
try_into_payload had two callers meaning different things by Schema::Proto — the sink SDK meant the payload's variant, runtime/src/source.rs:609 means the wire format a plugin sent. Split in d8a8ece: the sink rebuilds through Payload::try_from_schema, so all six variants keep theirs. Elasticsearch, Meilisearch and the ClickHouse string passthrough now take proto text as text; the JSON-only sinks still reject it, noted in Compatibility.
One correction: json_to_protobuf only falls back to Payload::Proto with no descriptor (proto_convert.rs:656) or non-object JSON (:285); with both it returns Payload::Raw(binary), which round-trips.
Regression tests in d8a8ece: the round-trip test no longer exempts Proto, and a new Elasticsearch integration test drives a proto_convert pipeline through a real runtime and container.
| // One `Schema` tag covers a whole FFI call, so messages are grouped into | ||
| // contiguous runs of the same payload variant and each run is sent on its | ||
| // own. Every decoder and transform in tree is deterministic per instance, so | ||
| // a run holds the entire batch in practice and there is one call. |
There was a problem hiding this comment.
The claim that an in-tree transform always yields one run is not quite true. ProtoConvert::json_to_protobuf() can return Raw for a successful schema conversion and Proto for a fallback on another message in the same batch. Could we remove the “one call” assertion and test a mixed-output batch?
There was a problem hiding this comment.
Agreed, the comment overreached. Dropped in d8a8ece; it now names ProtoConvert and claims only that a uniform batch stays one run.
json_to_protobuf returns Payload::Raw when encode_json_with_schema succeeds (proto_convert.rs:646) and Payload::Proto when it does not (:652), discriminated per message by whether the JSON is a top-level object (:285), so one instance with a descriptor alternates. It also contradicted the branch's second commit, which caps the per-run reservation for exactly that case.
The existing split test drove the mix with a stub transform. d8a8ece adds two using the real one: a proto_convert unit test pinning both variants from one instance, and a runtime test running ProtoConvert through process_messages asserting three runs tagged Raw, Proto, Raw with nothing lost.
| Payload::Json(_) => Schema::Json, | ||
| Payload::Raw(_) => Schema::Raw, | ||
| Payload::Text(_) => Schema::Text, | ||
| Payload::Proto(_) => Schema::Proto, |
There was a problem hiding this comment.
The PR promises that the tag describes the actual payload after transforms. That does not hold for Payload::Proto:
- proto_convert.rs#L627-L657 can return
Payload::Proto(json_string)when no descriptor is configured or schema encoding fails. - The new runtime logic tags its UTF-8 bytes as
Schema::Proto. - The receiver’s Schema::try_into_payload interprets those bytes as binary protobuf
Any, falling back toPayload::Raw, never reconstructingPayload::Proto.
There was a problem hiding this comment.
Confirmed, all three steps hold. Schema::try_into_payload served two callers that meant different things by Schema::Proto: the sink SDK meant the payload's variant, runtime/src/source.rs:609 means the wire format a plugin sent. Split in d8a8ece, so the sink rebuilds through Payload::try_from_schema and the source arm is untouched. Detail in @rohankumardubey's thread.
One addition to step 1: with a descriptor loaded, encode_json_with_schema also falls back on non-object JSON (proto_convert.rs:285), so one instance can emit both variants.
rohankumardubey
left a comment
rohankumardubey
left a comment
There was a problem hiding this comment.
@MarcusKainth left couple of comments to address. Thanks for contributing.
Schema::Proto means protobuf wire bytes when a source plugin sets it and a Payload::Proto string when the runtime tags a batch from the payload, so the sink SDK rebuilds through Payload::try_from_schema, an exact inverse of Payload::schema. Elasticsearch, Meilisearch and the ClickHouse string passthrough take proto text the way they take text. Also corrects the run-grouping comment, which claimed one FFI call per batch when ProtoConvert can return a different variant per message.
mlevkov
left a comment
mlevkov
left a comment
There was a problem hiding this comment.
Adversarial multi-agent review of d8a8ece9: four independent reviewers, then clean-room validators who saw only the raw claims and the diff, then tiebreaks on the three contested items. 10 of 22 initial claims were corrected and 1 was removed before anything reached this comment.
First, the parts that hold. The round trip is correct for all six Payload variants, the split of Schema::try_into_payload into two inverses is the right shape, and every line number in your reply to @rohankumardubey and @jiengup checks out. The earlier regression is genuinely fixed.
Three findings look like merge blockers, all the same shape: this PR establishes a remedy and applies it to three sinks out of six.
You added | Payload::Proto(text) to the Payload::Text arm in Elasticsearch, Meilisearch and the ClickHouse passthrough. build_json_body, build_row_binary_body, Doris, Delta and Iceberg did not get it, and all of them lose committed messages as a result. The offset commits at poll time (runtime/src/sink.rs:515), there is no replay and no dead-letter queue, so a failed or skipped batch is gone. iggy_connector_messages_processed_total counts the dropped rows and iggy_connector_errors_total stays flat, so the dashboard reads the same before and after the upgrade. Details inline on body.rs and doris_sink/src/lib.rs.
The Compatibility section is right that this surfaces a misconfiguration. The part that does not follow is the surfacing: on master these pipelines delivered correct rows, and after the upgrade they deliver none, with metrics reporting success.
Two corrections to the section itself: flat_buffer and proto streams also change tag and appear in neither the affected nor the unaffected list, and the descriptor-less case is not the only trigger. encode_json_with_schema returns Err for any non-object top-level JSON (proto_convert.rs:285), so a fully configured pipeline takes the fallback on every array, string, number, boolean and null message.
Also worth a look, not blocking:
- Four test gaps. The round-trip test compares only the variant, the per-run subtraction has no failing-run test, the new Elasticsearch test asserts the flattened shape, and the
Anytest does not check thattype_urlandvaluesurvived. Inline where they anchor. surrealdb_sink/src/lib.rs:838persistsmessages_metadata.schemaintoiggy_schema, so new rows on an avro stream now sayjson. Operator-visible, worth a README line.http_sink/src/lib.rs:1197says a nonzero return records no processed messages for the batch. The runtime now subtracts only the failed run..claude/skills/connector-sdk/SKILL.md:82points item 12 at a schema list incore/connectors/README.md. That file has none; the list is insdk/README.md.- Pre-existing, exposed rather than caused:
avro_convert.rs:125builds an encoder per message,proto_convert.rs:630serializes a document it then drops,proto_convert.rs:296can produce an emptyPayload::Rawthat now gets written, andelasticsearch_sink/src/lib.rs:373clones and parses per Raw message.
Simplifications, all optional: runtime/src/sink.rs:606 can use one flat Vec with run offsets instead of a Vec per run (postcard writes a borrowed slice and an owned Vec identically, so the FFI bytes are unchanged); sdk/src/lib.rs:294 can delegate five of six arms to Payload::try_from_schema, though it needs the variants named rather than a _ arm or it removes the exhaustiveness check, and mut value then trips -D warnings; runtime/src/sink.rs:751 can accumulate instead of subtracting. At flatbuffer_convert.rs:146 I would keep the Err return rather than unreachable!(), since the error drops one message and a panic kills the consumer task.
| 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. |
There was a problem hiding this 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.
| // 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. |
There was a problem hiding this 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.
| // 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(); |
There was a problem hiding this 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.
|
|
||
| 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. |
There was a problem hiding this 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.
| 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( |
There was a problem hiding this 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.
| 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([ |
There was a problem hiding this 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.
|
|
||
| for hit in &search_result.hits.hits { | ||
| assert_eq!( | ||
| hit.source.get("data_type").and_then(|v| v.as_str()), |
There was a problem hiding this 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.
| 0 | ||
| }; | ||
| ffi_elapsed += ffi_start.elapsed(); | ||
| if result != 0 { |
There was a problem hiding this 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.
| let rebuilt = Payload::try_from_schema(schema, bytes) | ||
| .unwrap_or_else(|error| panic!("failed to rebuild {schema} payload: {error}")); | ||
|
|
||
| assert_eq!( |
There was a problem hiding this 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.
| RuntimeError::FailedToSerializeMessagesMetadata | ||
| })?; | ||
| let mut ffi_elapsed = Duration::ZERO; | ||
| for (schema, run) in runs { |
There was a problem hiding this 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.
hubcio
added
S-waiting-on-author
5 participants
Which issue does this PR address?
Closes #4053
Rationale
Sinks configured
avro,protoorflatbufferreceive the wrongPayloadvariant. JSON-only sinks drop the batch after the offset has already been committed, so those messages are lost with no redelivery, and sinks that accept the raw variants store base64 of JSON text as though it were Avro bytes.What changed?
The runtime tagged each batch with the decoder's schema, which names the wire format a decoder reads rather than the variant it returns. All three decoders extract to
Payload::Jsonunder the configuration the runtime gives them, and the SDK rebuilds the payload from that tag alone, so a sink was handed aPayload::Avroholding JSON.The tag now comes from the payload itself through
Payload::schema(), read after transforms run. OneSchemacovers a whole FFI call, so messages are grouped into contiguous runs of the same variant, and a uniform batch stays one run and oneconsume()call. The new integration test coversavroonly, becauseStreamConsumerConfighas no schema configuration for flatbuffer or proto.Compatibility
S3, HTTP and SurrealDB will write a JSON document for avro streams instead of base64 of JSON text labelled as Avro bytes. That output is wrong today, so this is a correction, but anyone reading it will see the shape change. Streams configured
json,textorraware unaffected, because the decoder's schema and the payload's variant already agree.Payload::Protonow reaches a sink, which it never did before:Schema::try_into_payloadrebuilt that tag asPayload::Rawor, for validAnywire bytes,Payload::Json. Aproto_convertpipeline that falls back to proto text therefore lands differently. Elasticsearch, Meilisearch and the ClickHouse string passthrough now take it as text, alongside Quickwit, SurrealDB and S3, which already did; the HTTP sink already had aPayload::Protoarm but base64-encodes it rather than passing it through. ClickHouse's JSONEachRow and RowBinary modes now skip those rows and return success, and Delta, Doris and Iceberg fail the batch. On aschema = "json"stream the same payload previously arrived asPayload::Json, because the fallback text is the serialised JSON document, and was written. Those pipelines were relying on a mislabelled payload being silently reinterpreted, so this surfaces a configuration problem rather than creating one, but it is a behaviour change on upgrade.Local Execution
cargo fmt,cargo sort --no-format, Clippy with all features and all targets,taplo,markdownlint,shellcheck, license headers,typosandcargo macheteall pass. 232 connectors runtime and 204 SDK unit tests pass, along with the newschema_taggingintegration test against a real server and runtime.prek runpasses on the files this branch changes, covering markdownlint, license headers, version consistency, trailing whitespace and newline, binary artifacts, typos, skills, taplo,cargo fmtandcargo sort.AI Usage
Protoround-trip. The integration test sends real Avro datums through a running server and connectors runtime and asserts the sink is handedPayload::Json. The local checks listed above were run.