rohankumardubey commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4036453921
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -157,6 +157,26 @@ pub enum Payload {
}
impl Payload {
+ /// The `Schema` describing this payload's variant.
+ ///
+ /// Not the same thing as `StreamDecoder::schema`, which names the wire
+ /// format a decoder reads rather than the variant it hands back: the Avro
+ /// and FlatBuffer decoders return `Payload::Json` whenever
`extract_as_json`
+ /// is set, and the Proto decoder returns `Payload::Json` or `Payload::Raw`
+ /// depending on the path it takes. A transform may change the variant
again
+ /// after that. Anything tagging a payload for transport has to read the
tag
+ /// off the payload it actually holds.
+ pub const fn schema(&self) -> Schema {
+ match self {
+ Payload::Json(_) => Schema::Json,
+ Payload::Raw(_) => Schema::Raw,
+ Payload::Text(_) => Schema::Text,
+ Payload::Proto(_) => Schema::Proto,
Review Comment:
Payload::Proto contains text, but tagging it as Schema::Proto does not
round-trip through the sink SDK: Schema::Proto::try_into_payload() reconstructs
it as Raw (or Json for a protobuf Any), never Proto. The existing JSON→Proto
transform can produce this text, so this change can cause a sink to receive Raw
where it previously received Json and silently skip the message. Could we
preserve the intended payload type across the FFI boundary and add a regression
test for that transform path?
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -599,8 +596,14 @@ async fn process_messages(
}
let decode_elapsed = decode_start.elapsed();
- let mut messages = Vec::with_capacity(decoded.len());
+ // One `Schema` tag covers a whole FFI call, so messages are grouped into
+ // contiguous runs of the same payload variant and each run is sent on its
+ // own. Every decoder and transform in tree is deterministic per instance,
so
+ // a run holds the entire batch in practice and there is one call.
Review Comment:
The claim that an in-tree transform always yields one run is not quite true.
`ProtoConvert::json_to_protobuf()` can return `Raw` for a successful schema
conversion and `Proto` for a fallback on another message in the same batch.
Could we remove the “one call” assertion and test a mixed-output batch?
--
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]
