fix(connectors): preserve transformed payload schema across FFI

[Standing-Man]

Problem

The runtime previously populated both MessagesMetadata.schema and RawMessages.schema from decoder.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 decoded Payload with a different
variant.

For example, consider a sink configured with a raw input stream and a transform that converts raw bytes to text:

Iggy message bytes
    |
    | decode with Schema::Raw
    v
Payload::Raw(b"hello")
    |
    | Raw-to-Text transform
    v
Payload::Text("hello")
    |
    | Payload::try_into_vec()
    v
b"hello"
    |
    | FFI metadata incorrectly contains Schema::Raw
    v
Payload::Raw(b"hello") received by the sink plugin

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 the
transformed bytes as Payload::Raw. The transform appeared to succeed inside
the 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:

Payload::Proto
    -> Protobuf-to-JSON transform
    -> JSON bytes
    -> FFI metadata still contains Schema::Proto
    -> plugin attempts to interpret JSON bytes as Protobuf

Depending on the decoder and sink implementation, this could result in:

• the sink receiving a different Payload variant than the transform produced;
• transformed data falling back to Payload::Raw;
• payload decoding errors;
• format-specific sink validation rejecting the message;
• messages being skipped even though the transform itself succeeded.

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:

RawMessages {
    schema: decoder.schema(),
    messages,
}

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 then
applies the configured transform chain before serializing the payload and
passing the batch to the sink plugin over FFI.

Format-conversion transforms can change the Payload variant. For example, a
message decoded as Payload::Raw can become Payload::Text, or a Protobuf
payload can become JSON. Before this change, the runtime serialized the
transformed payload bytes but still attached decoder.schema(), which
describes the input bytes before transformation.

The FFI boundary only carries serialized bytes and schema metadata. On the
plugin side, SinkContainer::consume reconstructs each Payload by calling
MessagesMetadata::schema.try_into_payload. A stale input schema therefore
caused the transformed bytes to be interpreted as the wrong payload type. A
Raw-to-Text transform could arrive as Payload::Raw, while format-sensitive
conversions such as Protobuf-to-JSON could be rejected or decoded incorrectly.

What changed?

Payload::schema() now returns the schema represented by the current payload
variant. 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, schema
discriminants, 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 are
preserved, and that the batch reports one processed message.

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.

[github-actions]

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.

[codecov]

Codecov Report

❌ Patch coverage is 89.42308% with 11 lines in your changes missing coverage. Please review.
✅ Project coverage is 51.60%. Comparing base (c0218f1) to head (54617ea).
⚠️ Report is 9 commits behind head on master.

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.99%   51.60%   -22.40%     
  Complexity      937      937               
=============================================
  Files          1301     1295        -6     
  Lines        147389   130496    -16893     
  Branches     122949   106357    -16592     
=============================================
- Hits         109057    67339    -41718     
- Misses        34861    60204    +25343     
+ Partials       3471     2953      -518     

Components	Coverage Δ
Rust Core	46.61% <89.42%> (-27.68%)	⬇️
Java SDK	62.44% <ø> (ø)
C# SDK	72.21% <ø> (+0.01%)	⬆️
Python SDK	92.17% <ø> (ø)
PHP SDK	84.29% <ø> (ø)
Node SDK	91.35% <ø> (ø)
Go SDK	42.28% <ø> (-0.44%)	⬇️

Files with missing lines	Coverage Δ
core/connectors/sdk/src/lib.rs	57.14% <66.66%> (+0.96%)	⬆️
core/connectors/runtime/src/sink.rs	81.42% <91.57%> (+2.52%)	⬆️

... and 393 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.

[Standing-Man]

Hi @hubcio, could you please take a look when you have chance? Thanks.

[ryerraguntla]

@Standing-Man - Could you tell issue it is trying to solve and any related issue?

[ryerraguntla]

/author

[Standing-Man]

Could you tell issue it is trying to solve and any related issue?

I’ve added more details to the PR description to better explain the problem this PR addresses.

/request-review @ryerraguntla