Skip to content

fix(connectors): preserve transformed payload schema across FFI#3669

Open
Standing-Man wants to merge 1 commit into
apache:masterfrom
Standing-Man:fix-ffi-payload-schema
Open

fix(connectors): preserve transformed payload schema across FFI#3669
Standing-Man wants to merge 1 commit into
apache:masterfrom
Standing-Man:fix-ffi-payload-schema

Conversation

@Standing-Man

Copy link
Copy Markdown
Contributor

Which issue does this PR address?

Closes #
Relates to #

Rationale

Sink transforms can change the payload format after a message has been decoded.
For example, a transform may convert a raw payload into text or a Protobuf
payload into JSON.

The runtime serialized the transformed payload bytes but continued to attach
the original decoder schema to the FFI batch. The sink plugin then reconstructed
the payload using that stale schema, which could cause transformed bytes to be
decoded as the wrong payload type or rejected entirely.

What changed?

The payload schema is now derived from the transformed Payload variant before
the message crosses the FFI boundary. The derived schema is written to both
MessagesMetadata and RawMessages, keeping the two existing FFI metadata
representations consistent.

Messages whose transformed schema differs from the rest of the batch are
rejected and counted as errors instead of being reinterpreted using an
incompatible schema. The FFI function signatures, #[repr(C)] layouts, and
postcard payload structures remain unchanged.

A regression test converts a raw payload to text and verifies inside the FFI
callback that both metadata structures carry Schema::Text.

Local Execution

  • Passed / not passed
    passed
  • Pre-commit hooks ran / not ran
    ran

AI Usage

  1. Tool: Codex
  2. Scope: Root-cause analysis, implementation, regression test, and PR wording.
  3. Verification: The affected SDK and runtime crates passed formatting, Clippy,
    unit tests, integration-style SDK tests, and representative plugin builds.
  4. I reviewed the changes and can explain every modified line.

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>
@github-actions

Copy link
Copy Markdown

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

  • /ready - back to S-waiting-on-review after addressing feedback
  • /author - flip to S-waiting-on-author while you finish changes
  • /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

@github-actions github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Jul 14, 2026
@codecov

codecov Bot commented Jul 14, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 89.42308% with 11 lines in your changes missing coverage. Please review.
✅ Project coverage is 45.48%. Comparing base (5a5bc9a) to head (c0f89f7).

Files with missing lines Patch % Lines
core/connectors/runtime/src/sink.rs 91.57% 7 Missing and 1 partial ⚠️
core/connectors/sdk/src/lib.rs 66.66% 3 Missing ⚠️
Additional details and impacted files
@@              Coverage Diff              @@
##             master    #3669       +/-   ##
=============================================
- Coverage     73.49%   45.48%   -28.01%     
  Complexity      937      937               
=============================================
  Files          1291     1289        -2     
  Lines        141568   125216    -16352     
  Branches     117128   100776    -16352     
=============================================
- Hits         104046    56956    -47090     
- Misses        34300    65611    +31311     
+ Partials       3222     2649      -573     
Components Coverage Δ
Rust Core 38.59% <89.42%> (-35.12%) ⬇️
Java SDK 62.44% <ø> (ø)
C# SDK 72.21% <ø> (ø)
Python SDK 92.17% <ø> (ø)
PHP SDK 84.29% <ø> (ø)
Node SDK 91.35% <ø> (ø)
Go SDK 42.28% <ø> (ø)
Files with missing lines Coverage Δ
core/connectors/sdk/src/lib.rs 56.12% <66.66%> (-0.06%) ⬇️
core/connectors/runtime/src/sink.rs 75.92% <91.57%> (-2.97%) ⬇️

... and 492 files with indirect coverage changes

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

S-waiting-on-review PR is waiting on a reviewer

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant