MarcusKainth opened a new pull request, #4204: URL: https://github.com/apache/iggy/pull/4204
## Which issue does this PR address? Closes #4053 ## Rationale Sinks configured `avro`, `proto` or `flatbuffer` receive the wrong `Payload` variant. 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::Json` under the configuration the runtime gives them, and the SDK rebuilds the payload from that tag alone, so a sink was handed a `Payload::Avro` holding JSON. The tag now comes from the payload itself through `Payload::schema()`, read after transforms run. One `Schema` covers a whole FFI call, so messages are grouped into contiguous runs of the same variant, and a uniform batch stays one run and one `consume()` call. The new integration test covers `avro` only, because `StreamConsumerConfig` has 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`, `text` or `raw` are unaffected, because the decoder's schema and the payload's variant already agree. ## Local Execution - Passed. `cargo fmt`, `cargo sort --no-format`, Clippy with all features and all targets, `taplo`, `markdownlint`, `shellcheck`, license headers, `typos` and `cargo machete` all pass. 232 connectors runtime and 204 SDK unit tests pass, along with the new `schema_tagging` integration test against a real server and runtime. - Pre-commit hooks ran. `prek run` passes on the files this branch changes, covering markdownlint, license headers, version consistency, trailing whitespace and newline, binary artifacts, typos, skills, taplo, `cargo fmt` and `cargo sort`. ## AI Usage 1. Claude Opus 5. 2. Review and rebase, not implementation. I wrote the code. Claude reviewed it afterwards and I applied its suggestions with my own reasoning. Claude also rebased the branch onto current master and resolved the conflicts against #4152. 3. The unit tests pin the schema tag per variant, the one-FFI-call-per-uniform-batch invariant and the `Proto` round-trip. The integration test sends real Avro datums through a running server and connectors runtime and asserts the sink is handed `Payload::Json`. The local checks listed above were run. 4. Yes. -- 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]
