connectors: batches are tagged with the decoder's schema, not the payload's, so avro/proto/flatbuffer reach sinks as the wrong Payload variant

[MarcusKainth]

Bug description

When a sink's [[streams]] entry sets schema = "avro", "proto" or "flatbuffer", the sink receives its messages as the wrong Payload variant.

The runtime tags each batch with the decoder's schema rather than the type the decoder returned. Those three decoders all return Payload::Json under the config the runtime gives them, so the tag and the bytes disagree. The SDK rebuilds the payload from the tag, so a sink is handed a Payload::Avro holding JSON.

What happens next depends on the sink. Sinks that only handle Payload::Json — ClickHouse, Delta, Elasticsearch, Meilisearch, Quickwit, Iceberg — log the message as an unsupported payload type and drop it. Sinks that accept the raw variants — S3, HTTP, SurrealDB — base64-encode the JSON text and store it as though it were Avro bytes.

Expected: the sink receives Payload::Json, because that is what the decoder produced.

Root cause

// core/connectors/runtime/src/sink.rs:359-363
let messages_metadata = MessagesMetadata {
    partition_id,
    current_offset,
    schema: decoder.schema(),   // Schema::Avro
};

The bytes sent alongside come from Payload::try_into_vec(), which for the Payload::Json the Avro decoder returned gives JSON (core/connectors/sdk/src/lib.rs:163-165). The same tag is put on RawMessages at :731. The SDK rebuilds from it (sdk/src/sink.rs:199, then lib.rs:263: Schema::Avro => Payload::Avro(value)).

No connector config avoids this. The runtime builds the Avro decoder with ..AvroConfig::default() (sink.rs:532-536), so extract_as_json cannot be turned off from a connector file, and Proto and FlatBuffer get Schema::decoder() defaults with no knob at all.

Sources are unaffected: a source plugin tags its own ProducedMessages.schema and the encoder takes the Payload directly (source.rs:430-435), with no rebuild step.

Which schemas are affected

schema	decoder returns	batch tagged	sink receives
json, text, raw	matching variant	matching	works
avro	Payload::Json (decoders/avro.rs:169)	Avro	Payload::Avro
flatbuffer	Payload::Json (decoders/flatbuffer.rs:109)	FlatBuffer	Payload::FlatBuffer
proto	Payload::Json, if the message is an Any (decoders/proto.rs:529)	Proto	Payload::Raw

The three that work are the ones whose decoder output matches its own schema(). Avro and FlatBuffer default to extract_as_json: true; Proto with no schema configured falls through to decode_as_any (proto.rs:264-265), which also returns JSON. Proto ends up as Payload::Raw rather than Payload::Proto because try_into_payload re-decodes the bytes as a prost_types::Any and falls back to Raw when that fails (lib.rs:260).

Impact

For the Json-only sinks the batch is lost without an error being raised. The ClickHouse sink ends up with an empty body and returns Ok(()) (sinks/clickhouse_sink/src/sink.rs:146-152), and offsets are committed with the poll request (AutoCommit::When(AutoCommitWhen::PollingMessages), runtime/src/sink.rs:522) — the SDK documents that mode as sending "the commit with the poll request itself, before your code sees the batch" (core/sdk/src/clients/consumer.rs:501). So the offsets advance before the sink runs, and restarting the connector does not redeliver.

The batch is still counted as consumed. A run of 100 messages logs Consumed 100 messages, and the connector reports Processed 0 messages. at close.

Related to #3950 and PR #3954, which bind the sink's return status and let a sink defer the offset commit. Neither covers this: the ClickHouse sink returns Ok(()) on an empty body, so even with offset_commit = after_consuming the batch would be committed as successfully written.

Suggested fix

Tag the batch with the payload's type rather than the decoder's. Changing each decoder's schema() would not be enough on its own: transforms run after decoding and can change the type again, and the Proto decoder returns either Json or Raw depending on the path it takes.

Adding a Payload::schema() and using it where MessagesMetadata and RawMessages are built (runtime/src/sink.rs:362 and :731) covers all three formats in one place. I can send a PR either way; say which you prefer.

Compatibility

The fix changes what three sinks write. S3, HTTP and SurrealDB match on Payload::Avro and base64-encode it, so an avro stream reaches them today as base64 of JSON text, stored as though it were Avro bytes. Nothing reports this. After the fix they receive Payload::Json and write a JSON document, so anyone consuming their current output would see it change shape. That output is corrupt.

The Json-only sinks go from dropping the batch to writing it correctly. Streams configured json, text or raw are unaffected, because the decoder's schema and the payload's type already agree and the tag does not move.

Two related gaps:

• StreamConsumerConfig only exposes avro_schema_json and avro_schema_path, so flatbuffer and proto cannot be given a schema through the runtime at all.
• Nothing exercises a non-JSON schema through the runtime's sink path, and all ten configs under runtime/example_config/connectors/ use schema = "json". A test across the decoders would catch this class of problem.

Affected area / component

Connectors

Deployment

Compiled from source

Versions

master @ 412014a — iggy-connectors 0.5.0-edge.6, iggy_connector_sdk 0.4.0-edge.3, iggy_connector_clickhouse_sink 0.2.0-edge.4

Hardware / environment

macOS arm64, local build; not hardware-dependent.

Sample code

type = "sink"
key = "clickhouse"
enabled = true
version = 0
name = "ClickHouse sink"
path = "target/release/libiggy_connector_clickhouse_sink"
verbose = true

[[streams]]
stream = "repro"
topics = ["avro"]
schema = "avro"
avro_schema_json = '{"type":"record","name":"Event","fields":[{"name":"id","type":"long"},{"name":"name","type":"string"}]}'
batch_length = 100
poll_interval = "5ms"
consumer_group = "repro-sink"

[plugin_config]
url = "http://localhost:8123"
database = "default"
table = "repro"
insert_format = "row_binary"

Producer — an ordinary IggyProducer, with each payload an Avro datum for that schema:

let schema = apache_avro::Schema::parse_str(SCHEMA)?;
let record = apache_avro::types::Value::Record(vec![
    ("id".to_owned(), apache_avro::types::Value::Long(id)),
    ("name".to_owned(), apache_avro::types::Value::String(format!("row-{id}"))),
]);
let datum = apache_avro::writer::datum::GenericDatumWriter::builder(&schema)
    .build()?
    .write_value_to_vec(record)?;
let message = IggyMessage::builder().payload(datum.into()).build()?;

Logs

100 Avro messages into a single-partition topic, ClickHouse sink in row_binary mode:

INFO  iggy_connectors::sink: Processing 100 messages for sink connector with ID: 1
ERROR connector: connector_target="iggy_connector_clickhouse_sink::body"
      RowBinary mode: skipping unsupported payload type at offset 0
ERROR ... at offset 1, 2, ... 99
ERROR connector: connector_target="iggy_connector_clickhouse_sink::sink"
      ClickHouse sink ID: 1 — no serialisable messages in batch of 100
INFO  iggy_connectors::sink: Consumed 100 messages in 675µs for sink connector with ID: 1
INFO  connector: ClickHouse sink ID: 1 closed. Processed 0 messages.

SELECT count() FROM default.repro returns 0.

Restarting the connector against the same consumer group redelivers nothing: no batch, no errors, still no rows. The offsets were committed at poll time, so the 100 messages cannot be recovered without resetting them by hand.

Tagging the batch from the payload's type rather than the decoder's makes the same configuration land all 100 rows with the expected values, which confirms the cause.

Iggy server config

Default server config; not involved.

Reproduction

1. Run an Iggy server and a ClickHouse instance with
   CREATE TABLE repro (id Int64, name String) ENGINE = MergeTree ORDER BY id.
2. Produce Avro-encoded records matching the schema above to repro/avro.
3. Run iggy-connectors with the sink config above.
4. The log fills with skipping unsupported payload type and the table stays
   empty.
5. Restart the connector: nothing is redelivered, and the table is still empty.

Contribution

• I'm willing to submit a pull request to fix this bug

Good first issue

• I think this could be a good first issue for a new contributor