fix(connectors): tag sink batches with the payload's schema - #4204
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.59% 68.63% -18.97%
Complexity 1575 1575
=============================================
Files 1284 1282 -2
Lines 224493 185387 -39106
Branches 187856 148750 -39106
=============================================
- Hits 196653 127246 -69407
- Misses 23127 53349 +30222
- Partials 4713 4792 +79
🚀 New features to boost your workflow:
|
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.
hubcio
added
S-waiting-on-author
…oad-schema-tagging
The meaning of Schema::Proto on the wire changed: the runtime tags a sink batch from the payload, so a proto tag names a Payload::Proto string rather than protobuf wire bytes. A plugin built against 0.4.0 still runs the old inverse and rebuilds such a run as Payload::Raw, and nothing at load time can catch that because iggy_sink_version reports the plugin's own crate version. The SDK is neither published nor tagged, so 0.5.0 is a compatibility marker rather than a release. The connector-sdk skill now treats a Schema variant that changes meaning as breaking.
A proto_convert transform with no descriptor, or one that cannot encode a message, hands the sink Payload::Proto holding the JSON it was given. On master that run was tagged json and reparsed, so the document sinks loaded it. Tagging from the payload made ClickHouse skip the rows and return success, Doris abort the poll, Delta fail the batch, Iceberg drop the rows, and Elasticsearch and Meilisearch flatten the document to one text field. The offset commits at poll time, so each of those was silent loss or a lossy index. Payload::json_document borrows a Json payload and parses a Proto one, so every document sink takes the document when there is one and keeps its existing handling when there is not: proto text that is not JSON still skips, aborts or indexes as text as before. The Elasticsearch integration test asserts the original fields rather than the flattened shape, and ClickHouse and Doris gain the same pipeline end to end. The sink round-trip test compares bytes as well as the variant.
Quickwit flattened Payload::Proto to a text wrapper before this branch, the same shape Elasticsearch and Meilisearch had. It now takes the document when the text is JSON and keeps the wrapper otherwise, so the document sinks agree on what proto text means.
Nothing recorded that a batch was split into several consume() calls, and processed_count started at the batch size and was debited per failed run. iggy_connector_sink_runs counts the calls, so runs per batch is that counter over the total-stage sample count, and the benchmark event carries the same number. processed_count accumulates in the success branch, which removes the underflow class outright. The ffi stage sample stays one per batch, summed over its runs, and the README says so. Tests cover one failing run in a split batch and every run failing.
The sink SDK's rebuild had no direct test; only an integration test crossed it. The container is now driven in process with postcard bytes, so a proto run arriving as Payload::Proto is asserted where the runtime cannot influence it, with json, text and raw runs as controls. The protobuf Any test asserts type_url and value rather than key presence.
Transforms run inside the runtime's consume task, so a panic on a conversion pair that validate_conversion did not accept would take the task down rather than one message. The pair cannot reach that arm today; it now returns the error validation would have returned.
One poll can arrive as several consume() calls that repeat the same current_offset, and an empty batch keeps the stream's configured schema. The sdk README limits the source_format rejection to flatbuffer_convert, the only transform with the guard, and states that chain order is undefined. The http sink's runtime note and the SurrealDB iggy_schema column are updated to match.
… HTTP The same shape as the document sinks: a proto_convert transform with no descriptor hands over the JSON it was given as proto text, and these three wrote it as a string or base64 where a batch tagged json used to give them the document. All three now read the document when the text is JSON. HTTP sends proto text that is not JSON as a plain string, the way it sends Text, since a text stream with proto_convert now delivers Proto where it delivered Text.
|
Thanks for the depth here. All twelve threads are addressed and the per-thread replies cite the commits. Three things I want on record at the top level. First, the evidence. Rather than argue the before and after, I ran the same test files against a
The rows that fail on Building that matrix surfaced two additions to the Compatibility section. A A second pass on my side after the fixes turned up the same defect in three more sinks and one metric detail, so for completeness:
Second, four attributions I would correct. RowBinary never had a failure signal to lose; the builder on Third, the SDK bump. Agreed and done, 0.4.0 to 0.5.0, with the rebuild requirement in the PR body, and the skill now treats a |
|
/ready |
github-actions
Bot
added
S-waiting-on-review
github-actions
Bot
added
S-waiting-on-author
…oad-schema-tagging
The same four lines rebuilding a Payload::Proto that holds JSON into a Payload::Json had been copied into five sinks, each under its own wording of the same comment. Payload::into_json_document states the rule once, next to json_document, so it lives with the type rather than with every sink that has to follow it. The Iceberg dynamic router now normalizes at the top of its routing loop rather than at each point of use, so the proto text behind a route field is no longer parsed once for routing and again when the data files are written.
The run-splitting test rebuilt the four retagged messages that split_batch already builds, while the two tests below it call the helper. An empty batch still reaching the sink as exactly one call is a contract the Sink::consume rustdoc is about to state, so it gets a test of its own rather than resting on a comment in the runtime.
A poll reaching the plugin as several consume() calls, each repeating one current_offset, was documented only in the sinks README, while the rustdoc on Sink::consume said a batch arrives every time one is received. A plugin author reading the trait was told the opposite of what the runtime does, so the contract now sits on consume and on MessagesMetadata as well. The SDK README listed the retry removals as the only changes that break out-of-tree plugins; the schema tag changing meaning belongs there too. The Meilisearch README described proto nowhere, leaving its readers to infer from "unsupported payload schemas are skipped" something the sink stopped doing.
|
The description now links #3669. All six threads are addressed in |
|
/ready |
github-actions
Bot
added
S-waiting-on-review
master moved 192 commits since this branch was cut and two of them land on the same code. apache#4152 already reads the FFI status a sink returns: it logs the failure and counts it, then still returns Ok. apache#4204 then split a batch into runs of contiguous payload variants, so one batch is now several FFI calls, each with its own status, and a failing run does not stop the ones after it. The single-call block here gives way to that loop. Halting still needs one verdict per batch, so the loop records the first non-zero status and the batch is committed only when the sink took every run. A partly accepted batch therefore commits nothing and its accepted prefix is redelivered. That keeps the semantics this branch already shipped, but it is not the tightest watermark available: runs are contiguous and in offset order, so the end of the last consecutively accepted run would commit more without losing anything. Left alone here because it is a behaviour change rather than a conflict. The config tests keep both sides, and to_sink_config became into_sink_config taking a ConnectorKey. Module ordering and a stray blank line in a toml are fixed, which clears the two one-line CI failures on this head. Nothing here answers the review on apache#3954.
6 participants
Which issue does this PR address?
Closes #4053
Prior art: #3669.
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
Three of the six stream
schemavalues change tag.avro,flat_bufferandprotoall extract to JSON under the runtime's default decoder settings (avro.rsL40,flatbuffer.rsL42,proto.rsL48), so a batch from any of them now arrives taggedjson. S3, HTTP and SurrealDB will write a JSON document for those streams instead of base64 of JSON text labelled as the wire format. That output was wrong, so this is a correction, but anyone reading it will see the shape change. Aprotostream into Doris improves outright: it failed every poll before and writes rows now.json,textandrawstreams are unaffected, because the decoder's schema and the payload's variant already agree.Payload::Protonow reaches a sink, which it never did before. Aproto_converttransform hands it over whenever it has no descriptor, and also whenever it has one and the message is not a top-level JSON object, becauseencode_json_with_schemareturnsErrfor any array or scalar (proto_convert.rsL285). On ajsonstream the same messages previously arrived asPayload::Json. Every sink that writes documents now reads aPayload::Protothat holds JSON as that document: ClickHouse (JSONEachRow and RowBinary), Doris, Delta, Iceberg, Elasticsearch, Meilisearch, Quickwit, S3, SurrealDB and HTTP. Proto text that is not JSON is handled the way each sink handlesText. RowBinary keeps failing a batch on a non-object document, as it did onmaster. Two smaller shifts remain: atextorrawstream withproto_convertnow hands the sinkPayload::Protowhere it handedTextorRaw, which changes nothing now that both are treated alike; and a descriptor-backedproto_converton ajsonstream, which the SDK dropped onmasterbecause thejsontag sent protobuf bytes through the JSON parser, now arrives asRaw.iggy_connector_sdkmoves from 0.4.0 to 0.5.0. No FFI signature orSchemavariant changed, but theprototag changed meaning on the wire. A plugin built against 0.4.0 is better off under this runtime onavro,flat_bufferandprotostreams, because it now receives ajsontag it can read, and worse off on one pipeline: ajsonstream withproto_convert, where it rebuilds theproto-tagged run asPayload::Rawwheremasterhanded itPayload::Json. Plugins built against 0.4.0 or earlier must be rebuilt.avro,flat_bufferandprotostreams now pay a per-message JSON parse inside the plugin. They avoided it before by handing the plugin a mislabelled payload it could not use.One poll can reach a plugin as several
consume()calls, one per contiguous payload variant, all repeating the samecurrent_offset.iggy_connector_sink_runs_totalcounts those calls; theffistage histogram stays one sample per batch, summed over its runs. SurrealDB'siggy_schemacolumn records the variant received, so rows written before this change onavro,flat_bufferandprotostreams disagree with new ones.Local Execution
cargo fmt,cargo sort --no-format, Clippy with all features and all targets on every touched crate and the integration crate,taplo,markdownlint, license headers, trailing whitespace and newline,typosandcargo macheteall pass. Unit tests pass for the runtime (236), the SDK (185) and the ten sinks touched, including the new tests for each. Theschema_tagging, Elasticsearch, ClickHouse and Dorisproto_textintegration tests pass against a real server, runtime and containers.masterworktree,d8a8eceand this branch. The three pipeline tests pass onmaster, fail ond8a8eceand pass here; the schema-tagging and sink container tests fail onmasterand pass on both later refs; everyjson,text,rawand source-path control passes on all three. The full table is in the review summary comment.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.