fix(connectors): preserve transformed payload schema across FFI - #3669
fix(connectors): preserve transformed payload schema across FFI#3669Standing-Man wants to merge 4 commits into
Conversation
Sink transforms can change the Payload variant after the input stream has been decoded, but the runtime forwarded decoder.schema() to sink plugins. Plugins then reconstructed the transformed bytes using the stale input schema. Derive the schema from transformed payloads and use it for both FFI metadata containers. Reject mixed-schema messages within a batch and cover Raw-to-Text conversion at the FFI boundary. Signed-off-by: StandingMan <jmtangcs@gmail.com>
|
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 #3669 +/- ##
=============================================
- Coverage 73.99% 51.87% -22.12%
Complexity 937 937
=============================================
Files 1301 1297 -4
Lines 147389 131852 -15537
Branches 122949 107713 -15236
=============================================
- Hits 109057 68399 -40658
- Misses 34861 60481 +25620
+ Partials 3471 2972 -499
🚀 New features to boost your workflow:
|
|
Hi @hubcio, could you please take a look when you have chance? Thanks. |
|
@Standing-Man - Could you tell issue it is trying to solve and any related issue? |
|
/author |
github-actions
Bot
added
S-waiting-on-author
I’ve added more details to the PR description to better explain the problem this PR addresses. /request-review @ryerraguntla |
github-actions
Bot
added
S-waiting-on-review
| None => vec![], | ||
| }; | ||
|
|
||
| if output_schema.is_some_and(|schema| schema != payload_schema) { |
There was a problem hiding this comment.
this drop path is reachable with in-tree transforms: ProtoConvert with source json, target proto, a descriptor and message_type set emits Payload::Raw when encode_json_with_schema succeeds and falls back to Payload::Proto when it fails (top-level json not an object). one batch mixing json objects with arrays/scalars produces mixed variants, and every message whose schema differs from the first surviving one gets dropped here - first-schema-wins, so a leading minority can drop the majority. with AutoCommit::When(PollingMessages) the offset is already committed, so the drop is permanent.
not a regression - pre-PR the same batch shipped under one stale label and was lost/corrupted plugin-side anyway - but this path has zero test coverage. worth a heterogeneous-batch test now, and longer-term splitting the batch per schema (one consume() per group) instead of dropping.
There was a problem hiding this comment.
I concur on test coverage, needs test coverage as much as possible
There was a problem hiding this comment.
Fixed. process_messages no longer applies a first-schema-wins policy. After transforms, messages are grouped into contiguous schema-homogeneous sub-batches, and each group is passed to consume() separately with matching MessagesMetadata::schema and RawMessages::schema.
I also added coverage through the SDK SinkContainer, so the test exercises postcard serialization, the FFI callback, SDK-side payload reconstruction, and the final Sink::consume() call. Additionally, transform errors are now counted only as errors; only Ok(None) increments the filtered-message metric.
cc @hubcio and @ryerraguntla.
github-actions
Bot
added
S-waiting-on-author
| 0 | ||
| } | ||
|
|
||
| #[tokio::test] |
There was a problem hiding this comment.
only homogeneous Raw→Text FFI test; no mixed-schema path. Fix: hetero-batch test (first-wins keep, mismatch error_count, FFI schema).
| continue; | ||
| }; | ||
|
|
||
| let payload_schema = message.payload.schema(); |
There was a problem hiding this comment.
sink.rs:677-706 — try_into_vec/headers before schema check; waste on mismatch. Fix: compare schema first.
There was a problem hiding this comment.
This mismatch path no longer exists. Messages with different transformed schemas are now preserved and placed into separate contiguous schema-homogeneous sub-batches, so payload and header serialization is required for every surviving message.
|
/author |
Signed-off-by: StandingMan <jmtangcs@gmail.com>
|
There are a few trade-offs to splitting a polled batch by transformed payload schema:
cc @hubcio and @ryerraguntla. |
|
/ready |
github-actions
Bot
added
S-waiting-on-review
| for batch in batches { | ||
| let messages_metadata = MessagesMetadata { | ||
| partition_id: messages_metadata.partition_id, | ||
| current_offset: messages_metadata.current_offset, | ||
| schema: batch.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(&batch).map_err(|error| { | ||
| error!("Failed to serialize messages for sink connector with ID: {plugin_id}. {error}"); | ||
| RuntimeError::FailedToSerializeRawMessages | ||
| })?; | ||
|
|
||
| let ffi_start = Instant::now(); | ||
| (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 ffi_start = Instant::now(); | ||
| (consume)( | ||
| plugin_id, | ||
| topic_meta.as_ptr(), | ||
| topic_meta.len(), | ||
| messages_meta.as_ptr(), | ||
| messages_meta.len(), | ||
| messages.as_ptr(), | ||
| messages.len(), | ||
| ); | ||
| ffi_elapsed += ffi_start.elapsed(); | ||
| } |
There was a problem hiding this comment.
This turns one poll into N consume() invocations, but the plugin-facing contract still says otherwise. Sink::consume in core/connectors/sdk/src/lib.rs documents it as "Invoked every time a batch of messages is received from the configured stream(s) and topic(s)", and core/connectors/sinks/README.md still describes the delivered format as fixed by the stream config.
The four trade-offs you listed in the thread are the contract now: current_offset repeated across sub-batches, no atomicity between them, no 1:1 poll-to-consume relationship. Out-of-tree plugins using (partition_id, current_offset) as an idempotency or dedup key will silently discard sub-batches 2..N. I went through the in-tree sinks and none of them break - all uses of current_offset are log-only and s3_sink keys off per-message offsets while buffering across calls - but there is no line anywhere in the tree that tells a plugin author this changed.
Please move those trade-offs into the Sink::consume doc comment and sinks/README.md.
Unrelated to the loop, but same theme: the transform-error accounting change in this PR (an error no longer also bumping iggy_connector_messages_filtered_total) is correct and brings the sink in line with source.rs, which already has the identical transform_failed flag. It also changes the meaning of an exported metric, and right now that only exists in a PR comment. Worth a line in the commit body.
| let processed_count = batches.iter().map(|batch| batch.messages.len()).sum(); | ||
| if batches.is_empty() { | ||
| batches.push(RawMessages { | ||
| schema: messages_metadata.schema, | ||
| messages: Vec::new(), | ||
| }); | ||
| } |
There was a problem hiding this comment.
Nothing reports how many sub-batches a poll produced.
Grouping is contiguous, so the worst case is one consume() per message. With the default batch_length = 1000 and a stream that alternates schemas - e.g. ProtoConvert encoding JSON objects successfully and falling back on arrays/scalars - a single poll becomes up to 1000 sequential FFI calls, each one an HTTP request or DB transaction on the plugin side, with no await point in this loop.
I'm not asking for a cap; a cap would reintroduce the drops this PR removes. But the fan-out has to be visible, otherwise the failure mode is a latency collapse with nothing pointing at sub-batching. batch_count alongside processed_count in benchmark::emit_sink_event, plus a debug! when batches.len() > 1, would be enough.
| assert_eq!( | ||
| captured | ||
| .iter() | ||
| .map(|batch| batch.metadata_schema) | ||
| .collect::<Vec<_>>(), | ||
| vec![Schema::Proto, Schema::Raw, Schema::Proto, Schema::Raw] | ||
| ); |
There was a problem hiding this comment.
This assertion locks in a mislabel that lives in ProtoConvert::json_to_protobuf: successful protobuf encoding returns Payload::Raw(binary) while the fallback returns Payload::Proto(json_string) - inverted. The PR forwards both faithfully, which is the right behaviour for this layer, but the two consequences are not equivalent:
- Success path (
Schema::Rawin this vector) is a genuine fix. Pre-PR this shippedSchema::Json, the SDK ran simd_json over protobuf binary, and the message was dropped plugin-side. - Fallback path (
Schema::Protoin this vector) is a regression. The SDK'sSchema::Protoarm triesprost_types::Any::decode, fails on JSON text, and yieldsPayload::Raw; pre-PR the plugin receivedPayload::Json.quickwit_sinkskips non-Jsonpayloads with awarn!and returnsOk(()), so underAutoCommit::When(PollingMessages)those messages are gone with the offset already committed.
Either fix the variant choice in proto_convert.rs here, or file a follow-up and add a comment at this assertion so the Schema::Proto entry is not read as intended behaviour by the next person touching it.
4 participants
Problem
The runtime previously populated both
MessagesMetadata.schemaandRawMessages.schemafromdecoder.schema().decoder.schema()describes the input format configured for the Iggy stream. It remains unchanged for the lifetime of the consumer. However, format-conversion transforms can replace the decodedPayloadwith a differentvariant.
For example, consider a sink configured with a raw input stream and a transform that converts raw bytes to text:
Although the transform produced
Payload::Text, converting it back into bytes removed the Rust enum variant information. The only information available to the plugin for reconstructing the payload was the schema in the FFI metadata.Because the runtime still sent
Schema::Raw, the plugin reconstructed thetransformed bytes as
Payload::Raw. The transform appeared to succeed insidethe runtime, but its output type was lost when the message crossed the FFI
boundary.
The same problem is more serious for schema-sensitive conversions. For example:
Depending on the decoder and sink implementation, this could result in:
The transformed bytes were serialized correctly. The problem was that the runtime attached stale type information to those bytes.
Root cause
The runtime treated the configured input schema as if it were also the output
schema of the transform chain:
That assumption only holds for transforms that preserve the payload format. It is invalid for ProtoConvert, AvroConvert, FlatBufferConvert, and any other transform that changes the Payload variant. The runtime already had the correct type information in the transformed Payload, but it discarded that information when calling
Payload::try_into_vec().Rationale
The sink runtime decodes each Iggy message according to the schema configured
for the input stream and represents the decoded value as a
Payload. It thenapplies the configured transform chain before serializing the payload and
passing the batch to the sink plugin over FFI.
Format-conversion transforms can change the
Payloadvariant. For example, amessage decoded as
Payload::Rawcan becomePayload::Text, or a Protobufpayload can become JSON. Before this change, the runtime serialized the
transformed payload bytes but still attached
decoder.schema(), whichdescribes the input bytes before transformation.
The FFI boundary only carries serialized bytes and schema metadata. On the
plugin side,
SinkContainer::consumereconstructs eachPayloadby callingMessagesMetadata::schema.try_into_payload. A stale input schema thereforecaused the transformed bytes to be interpreted as the wrong payload type. A
Raw-to-Text transform could arrive as
Payload::Raw, while format-sensitiveconversions such as Protobuf-to-JSON could be rejected or decoded incorrectly.
What changed?
Payload::schema()now returns the schema represented by the current payloadvariant. The sink runtime reads that schema after all transforms have completed
and before the payload is converted back into bytes.
The transformed output schema is written to both existing FFI metadata
containers:
MessagesMetadata.schema, which the sink SDK uses to reconstruct payloads.RawMessages.schema, which keeps the serialized batch metadata consistent.A sink batch must have one output schema because the existing FFI batch format
stores schema once per batch. If a transformed message has a different schema
from the messages already accepted into the batch, the runtime logs the
mismatch, increments the connector error counter, and skips that message rather
than decoding it later with an incompatible schema.
If every message is filtered or rejected, the batch retains the original input
schema as a safe fallback.
This keeps the fix compatible with existing plugins. It does not change FFI
function signatures,
#[repr(C)]layouts, postcard payload structures, schemadiscriminants, or connector configuration.
A regression test exercises the complete runtime-to-plugin boundary with a
Raw-to-Text transform. The FFI callback deserializes both metadata structures
and verifies that they carry
Schema::Text, that the transformed bytes arepreserved, and that the batch reports one processed message.
Local Execution
passed
ran
AI Usage
unit tests, integration-style SDK tests, and representative plugin builds.