You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/22 10:31:02 UTC
(iggy) branch master updated: fix(connectors): tag sink batches with the payload's schema (#4204)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7bae8ec84 fix(connectors): tag sink batches with the payload's schema (#4204)
7bae8ec84 is described below
commit 7bae8ec84d42103c27064535b14b4000680e717f
Author: Marcus Kainth <ma...@kainth.dev>
AuthorDate: Tue Sep 22 11:30:57 2026 +0100
fix(connectors): tag sink batches with the payload's schema (#4204)
---
.claude/skills/connector-runtime/SKILL.md | 5 +-
.claude/skills/connector-sdk/SKILL.md | 17 +-
.claude/skills/connector-sink/SKILL.md | 2 +-
Cargo.lock | 8 +-
Cargo.toml | 2 +-
core/connectors/runtime/Cargo.toml | 4 +
core/connectors/runtime/README.md | 5 +-
core/connectors/runtime/src/benchmark.rs | 5 +-
core/connectors/runtime/src/metrics.rs | 34 +-
core/connectors/runtime/src/sink.rs | 669 +++++++++++++++++++--
core/connectors/sdk/Cargo.toml | 2 +-
core/connectors/sdk/README.md | 15 +-
core/connectors/sdk/src/lib.rs | 334 +++++++++-
core/connectors/sdk/src/sink.rs | 149 ++++-
.../sdk/src/transforms/flatbuffer_convert.rs | 23 +-
.../connectors/sdk/src/transforms/proto_convert.rs | 65 +-
core/connectors/sinks/README.md | 8 +
core/connectors/sinks/clickhouse_sink/README.md | 8 +-
core/connectors/sinks/clickhouse_sink/src/body.rs | 110 +++-
core/connectors/sinks/delta_sink/Cargo.toml | 3 +
core/connectors/sinks/delta_sink/src/sink.rs | 81 ++-
core/connectors/sinks/doris_sink/README.md | 3 +-
core/connectors/sinks/doris_sink/src/lib.rs | 87 ++-
.../connectors/sinks/elasticsearch_sink/src/lib.rs | 80 ++-
core/connectors/sinks/http_sink/README.md | 3 +-
core/connectors/sinks/http_sink/src/lib.rs | 37 +-
.../iceberg_sink/src/router/dynamic_router.rs | 58 +-
.../sinks/iceberg_sink/src/router/mod.rs | 43 +-
.../connectors/sinks/influxdb_sink/dependencies.md | 2 +-
core/connectors/sinks/meilisearch_sink/README.md | 13 +-
core/connectors/sinks/meilisearch_sink/src/lib.rs | 38 +-
core/connectors/sinks/quickwit_sink/README.md | 2 +-
core/connectors/sinks/quickwit_sink/src/lib.rs | 22 +-
core/connectors/sinks/s3_sink/README.md | 2 +-
core/connectors/sinks/s3_sink/src/formatter.rs | 28 +-
core/connectors/sinks/surrealdb_sink/README.md | 11 +-
core/connectors/sinks/surrealdb_sink/src/lib.rs | 17 +
.../sources/influxdb_source/dependencies.md | 2 +-
core/integration/Cargo.toml | 1 +
.../integration/tests/connectors/clickhouse/mod.rs | 2 +
.../tests/connectors/clickhouse/proto_text.rs | 113 ++++
.../tests/connectors/clickhouse/proto_text.toml | 20 +
.../proto_text_config/clickhouse_sink.toml | 67 +++
core/integration/tests/connectors/doris/mod.rs | 1 +
.../tests/connectors/doris/proto_text.rs | 77 +++
.../tests/connectors/doris/proto_text.toml | 20 +
.../doris/proto_text_config/doris_sink.toml | 64 ++
.../tests/connectors/elasticsearch/mod.rs | 1 +
.../tests/connectors/elasticsearch/proto_text.rs | 126 ++++
.../tests/connectors/elasticsearch/proto_text.toml | 20 +
.../proto_text_config/elasticsearch_sink.toml | 62 ++
core/integration/tests/connectors/runtime/mod.rs | 1 +
.../tests/connectors/runtime/schema_tagging.rs | 134 +++++
.../tests/connectors/runtime/schema_tagging.toml | 20 +
.../runtime/schema_tagging_config/stdout_sink.toml | 36 ++
55 files changed, 2533 insertions(+), 229 deletions(-)
diff --git a/.claude/skills/connector-runtime/SKILL.md b/.claude/skills/connector-runtime/SKILL.md
index cee028180..8fb2695a6 100644
--- a/.claude/skills/connector-runtime/SKILL.md
+++ b/.claude/skills/connector-runtime/SKILL.md
@@ -213,8 +213,9 @@ All families labeled by `connector_key` + `connector_type` (histogram adds `stag
- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total` and `iggy_connector_errors_total`. These are the *rendered* names; each is registered without the `_total`, which the OpenMetrics encoder appends.
- `messages_filtered_total` - intentional drops via transform `Ok(None)`.
- - `errors_total` - unexpected drops (decode/encode/build failure, missing field, ...) + batch-level failures.
-- **Histograms**: `iggy_connector_stage_duration_seconds{stage}` (snake_case stage labels - `prepare`, `ffi`, `decode`, `iggy_send`, `state_save`, `total`). Buckets `STAGE_BUCKETS_SECONDS`. Always populated regardless of any flag. Scraped at `/metrics` when `[http.metrics] enabled = true`.
+ - `errors_total` - unexpected drops (decode/encode/build failure, missing field, ...) + batch-level failures, one per failed run.
+ - `sink_runs_total` - FFI `consume()` calls for sink batches, one per contiguous payload-variant run. Divide by `stage_duration_seconds_count{stage="total"}` for runs per batch.
+- **Histograms**: `iggy_connector_stage_duration_seconds{stage}` (snake_case stage labels - `prepare`, `ffi`, `decode`, `iggy_send`, `state_save`, `total`). Buckets `STAGE_BUCKETS_SECONDS`. Always populated regardless of any flag. One sample per batch for every stage; `ffi` is summed over the batch's runs. Scraped at `/metrics` when `[http.metrics] enabled = true`.
- **Gauges**: `iggy_connectors_{sources,sinks}_{total,running}`.
`/stats` JSON surface mirrors counters per-connector via `ConnectorStats` (`sdk/api.rs`): `messages_filtered`, `errors`, kind-specific counters.
diff --git a/.claude/skills/connector-sdk/SKILL.md b/.claude/skills/connector-sdk/SKILL.md
index 7d00fd484..8ad3a2791 100644
--- a/.claude/skills/connector-sdk/SKILL.md
+++ b/.claude/skills/connector-sdk/SKILL.md
@@ -63,6 +63,7 @@ sdk/src/
5. **`simd_json::OwnedValue`** for JSON payloads, not `serde_json::Value`. Use `convert::owned_value_to_serde_json` as a bridge when interop is required.
6. **`BTreeMap`** for headers - deterministic ordering. Never `HashMap` on the wire.
7. **No breaking changes** to `Sink`/`Source`/`StreamDecoder`/`StreamEncoder`/`Transform` trait signatures without coordinating with all in-tree plugins in the same PR.
+8. **A `Schema` variant that changes meaning is as breaking as a new variant.** The discriminant crosses FFI unchanged, so a plugin built against the old SDK keeps decoding it the old way and misreads every batch with no error anywhere. Same process as a trait change: bump the SDK minor version, update every in-tree plugin in the same PR, and state in the PR description that plugins built against the previous SDK must be rebuilt.
## Adding a new `Schema` variant
@@ -72,13 +73,15 @@ The `Schema` enum (in `lib.rs`) is `#[repr(C)]` - it crosses FFI. Touch points t
2. `Payload` enum: add a matching variant carrying the deserialized form.
3. `Payload::try_into_vec` - consuming bytes-out path.
4. `Payload::try_to_bytes` - **borrowing** bytes-out path. For non-trivial payloads (parsed trees), implement a no-clone serialization, not a `clone() + serialize`. See the `Payload::Json` arm for the canonical optimization.
-5. `Payload::Display`.
-6. `Schema::try_into_payload` - bytes → `Payload`.
-7. `Schema::decoder()` - factory returning `Arc<dyn StreamDecoder>`.
-8. `Schema::encoder()` - factory returning `Arc<dyn StreamEncoder>`.
-9. New `decoders/<name>.rs` and `encoders/<name>.rs`.
-10. Update `sdk/README.md` and `core/connectors/README.md` schema list.
-11. Tests: round-trip encode/decode, error paths.
+5. `Payload::schema()` - variant → `Schema`. The runtime tags every sink batch from this, so a missing arm will not compile but a wrong one silently mislabels the batch.
+6. `Payload::Display`.
+7. `Schema::try_into_payload` - bytes → `Payload`, for a tag naming a wire format (the source path).
+8. `Payload::try_from_schema` - bytes → `Payload`, for a tag that came from `Payload::schema()` (the sink path). The exact inverse of item 5.
+9. `Schema::decoder()` - factory returning `Arc<dyn StreamDecoder>`.
+10. `Schema::encoder()` - factory returning `Arc<dyn StreamEncoder>`.
+11. New `decoders/<name>.rs` and `encoders/<name>.rs`.
+12. Update the schema list in `sdk/README.md`.
+13. Tests: round-trip encode/decode, error paths. `Payload::schema()` must round-trip through `Payload::try_from_schema`, with no exceptions: every variant survives that trip and the test is table-driven over all of them.
Miss any of these → silent failures at runtime (typically `Error::InvalidPayloadType` or surprising decoder behavior in plugin code).
diff --git a/.claude/skills/connector-sink/SKILL.md b/.claude/skills/connector-sink/SKILL.md
index 207e55354..d13568282 100644
--- a/.claude/skills/connector-sink/SKILL.md
+++ b/.claude/skills/connector-sink/SKILL.md
@@ -54,7 +54,7 @@ for getting them to the external system reliably and efficiently.
### Payloads (efficiency-critical)
-- Dispatch on `messages_metadata.schema` and the `Payload` variant. Handle `Json`, `Raw`, `Text` at minimum. Unsupported -> `Error::InvalidPayloadType` or fall back to base64 (`elasticsearch_sink`).
+- Dispatch on the `Payload` variant. Handle `Json`, `Raw`, `Text` and `Proto` at minimum. `Proto` holds text that is a JSON document when `proto_convert` had no descriptor or could not encode, and arbitrary text otherwise: call `Payload::json_document()` first and fall back to the text handling when it returns `None`. Unsupported -> `Error::InvalidPayloadType` or fall back to base64 (`elasticsearch_sink`).
- For `Payload::Json`, call `try_to_bytes(&self)` - single-pass serialization without cloning the `OwnedValue` tree (`sdk/src/lib.rs::Payload::try_to_bytes`). Never `payload.clone().try_into_vec()`.
- To take a payload out of `ConsumedMessage` without cloning: `std::mem::replace(&mut message.payload, Payload::Raw(vec![]))`. Standard pattern across `http_sink::send_individual` / `send_ndjson` / `send_json_array`.
- Pre-allocate per-batch buffers with `Vec::with_capacity(messages.len())` (or `* factor` for serialized bodies). Used across `postgres_sink`, `mongodb_sink`, `quickwit_sink`, `elasticsearch_sink`, `http_sink`.
diff --git a/Cargo.lock b/Cargo.lock
index 43b1eeb33..c0a533ae0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6957,6 +6957,7 @@ dependencies = [
name = "iggy-connectors"
version = "0.5.0"
dependencies = [
+ "apache-avro 0.22.0",
"async-trait",
"axum",
"axum-server",
@@ -6984,6 +6985,9 @@ dependencies = [
"opentelemetry_sdk",
"postcard",
"prometheus-client",
+ "prost",
+ "prost-types",
+ "protox-parse",
"reqwest 0.13.5",
"reqwest-middleware",
"reqwest-retry",
@@ -7143,6 +7147,7 @@ dependencies = [
"secrecy",
"serde",
"serde_json",
+ "simd-json",
"tokio",
"tracing",
"url",
@@ -7496,7 +7501,7 @@ dependencies = [
[[package]]
name = "iggy_connector_sdk"
-version = "0.4.0"
+version = "0.5.0"
dependencies = [
"anyhow",
"apache-avro 0.22.0",
@@ -7764,6 +7769,7 @@ checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02"
name = "integration"
version = "0.0.1"
dependencies = [
+ "apache-avro 0.22.0",
"arrow",
"assert_cmd",
"async-trait",
diff --git a/Cargo.toml b/Cargo.toml
index 3a8243d89..e2154bdb2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -207,7 +207,7 @@ iggy-gateway-kafka = { path = "gateways/kafka" }
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0" }
iggy_common = { path = "core/common", version = "0.11.0" }
iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" }
-iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0" }
+iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.0" }
indexmap = "2.14.2"
integration = { path = "core/integration" }
ipnet = "2.12.2"
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 49fe7dea3..f871d76e1 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -80,6 +80,10 @@ tracing-subscriber = { workspace = true, features = ["json"] }
uuid = { workspace = true }
[dev-dependencies]
+apache-avro = { workspace = true }
+prost = { workspace = true }
+prost-types = { workspace = true }
+protox-parse = { workspace = true }
tempfile = { workspace = true }
wiremock = { workspace = true }
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
index eb907521e..580b7793e 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -381,7 +381,7 @@ benchmark = true
Emitted fields:
-- **sink**: `connector_key`, `stream`, `topic`, `partition_id`, `current_offset`, `batch_size`, `processed_count`, `decode_us`, `prepare_us`, `ffi_us`, `total_us`
+- **sink**: `connector_key`, `stream`, `topic`, `partition_id`, `current_offset`, `batch_size`, `processed_count`, `runs`, `decode_us`, `prepare_us`, `ffi_us`, `total_us`. `runs` is the number of `consume()` calls the batch was split into, one per contiguous payload variant, and `ffi_us` is summed over all of them.
- **source**: `connector_key`, `stream`, `topic`, `batch_size`, `sent_count`, `decode_us`, `prepare_us`, `iggy_send_us`, `state_saved`, `state_save_us`, `total_us` (`state_save_us` is 0 when `state_saved` is false)
Filter the stream via `RUST_LOG=iggy_connectors::benchmark=info`. The corresponding stage durations are also recorded in the `iggy_connector_stage_duration_seconds` histogram regardless of this flag, so Prometheus dashboards remain available without enabling text events.
@@ -405,6 +405,7 @@ The runtime exposes Prometheus-compatible metrics via the `/metrics` endpoint wh
- `iggy_connector_messages_processed_total`: Messages processed and sent to sink plugin
- `iggy_connector_messages_filtered_total`: Messages intentionally dropped by transforms returning `Ok(None)`
- `iggy_connector_errors_total`: Errors encountered
+- `iggy_connector_sink_runs_total`: FFI `consume()` calls made for sink batches. A batch is split into one call per contiguous payload variant, so this counter over `iggy_connector_stage_duration_seconds_count{stage="total"}` is the average number of runs per batch, and it equals that count when no batch splits
### Stage Duration Histograms (labeled with `connector_key`, `connector_type`, `stage`)
@@ -412,6 +413,8 @@ The runtime exposes Prometheus-compatible metrics via the `/metrics` endpoint wh
Stage label values (snake_case): `decode`, `prepare`, `total` on both sides; `ffi` on sinks; `iggy_send`, `state_save` on sources. The `connector_type` label is also snake_case (`source` / `sink`). Histograms are always exposed (independent of `benchmark`); buckets cover 50us to 5s.
+Every stage records one sample per batch. The `ffi` sample is the summed duration of every `consume()` call the batch was split into, so a batch that split into several runs still counts once; `iggy_connector_sink_runs_total` shows how often that happens.
+
**Cardinality:** each `(connector_key, stage)` pair yields ~16 series (13 explicit buckets + `+Inf` + `_sum` + `_count`), times 3 stages per sink or 5 per source, times the number of distinct `connector_key` values. `connector_key` comes from operator config, so keep it bounded. Do not template per-user or per-request keys into it or the time-series database will blow up.
## Stats
diff --git a/core/connectors/runtime/src/benchmark.rs b/core/connectors/runtime/src/benchmark.rs
index 8bef21f01..6dc564543 100644
--- a/core/connectors/runtime/src/benchmark.rs
+++ b/core/connectors/runtime/src/benchmark.rs
@@ -36,6 +36,7 @@ pub fn emit_sink_event(
current_offset: u64,
batch_size: usize,
processed_count: usize,
+ runs: usize,
decode_us: u64,
prepare_us: u64,
ffi_us: u64,
@@ -51,6 +52,7 @@ pub fn emit_sink_event(
current_offset = current_offset,
batch_size = batch_size,
processed_count = processed_count,
+ runs = runs,
decode_us = decode_us,
prepare_us = prepare_us,
ffi_us = ffi_us,
@@ -185,7 +187,7 @@ mod tests {
fn given_sink_event_emitted_when_captured_should_contain_all_fields() {
let events = capture(|| {
emit_sink_event(
- "postgres", "qw", "records", 0, 100, 50, 50, 120, 234, 1456, 1700,
+ "postgres", "qw", "records", 0, 100, 50, 50, 2, 120, 234, 1456, 1700,
);
});
let event = events
@@ -198,6 +200,7 @@ mod tests {
assert_eq!(event.fields.get("topic").unwrap(), "records");
assert_eq!(event.fields.get("batch_size").unwrap(), "50");
assert_eq!(event.fields.get("processed_count").unwrap(), "50");
+ assert_eq!(event.fields.get("runs").unwrap(), "2");
assert_eq!(event.fields.get("decode_us").unwrap(), "120");
assert_eq!(event.fields.get("prepare_us").unwrap(), "234");
assert_eq!(event.fields.get("ffi_us").unwrap(), "1456");
diff --git a/core/connectors/runtime/src/metrics.rs b/core/connectors/runtime/src/metrics.rs
index 5cc105859..9170c15c1 100644
--- a/core/connectors/runtime/src/metrics.rs
+++ b/core/connectors/runtime/src/metrics.rs
@@ -168,6 +168,7 @@ pub struct Metrics {
messages_processed: Family<ConnectorLabels, Counter>,
messages_filtered: Family<ConnectorLabels, Counter>,
errors: Family<ConnectorLabels, Counter>,
+ sink_runs: Family<ConnectorLabels, Counter>,
stage_duration_seconds: Family<StageLabels, Histogram, fn() -> Histogram>,
}
@@ -185,6 +186,7 @@ impl Metrics {
let messages_processed = Family::<ConnectorLabels, Counter>::default();
let messages_filtered = Family::<ConnectorLabels, Counter>::default();
let errors = Family::<ConnectorLabels, Counter>::default();
+ let sink_runs = Family::<ConnectorLabels, Counter>::default();
let stage_duration_seconds: Family<StageLabels, Histogram, fn() -> Histogram> =
Family::new_with_constructor(stage_histogram);
@@ -240,6 +242,13 @@ impl Metrics {
"Errors encountered",
errors.clone(),
);
+ // One batch is one `stage_duration_seconds{stage="total"}` sample, so
+ // runs per batch is this counter over that sample count.
+ registry.register(
+ "iggy_connector_sink_runs",
+ "FFI consume() calls made for sink batches, one per contiguous payload-variant run",
+ sink_runs.clone(),
+ );
registry.register(
"iggy_connector_stage_duration_seconds",
"Per-batch processing stage duration in seconds",
@@ -258,6 +267,7 @@ impl Metrics {
messages_processed,
messages_filtered,
errors,
+ sink_runs,
stage_duration_seconds,
}
}
@@ -434,6 +444,10 @@ impl Metrics {
}
}
+ pub fn inc_sink_runs_with_labels(&self, labels: &ConnectorLabels, count: u64) {
+ self.sink_runs.get_or_create(labels).inc_by(count);
+ }
+
/// Owned `errors` counter (Arc-shared atomic) for lookup-free hot-path increments.
pub fn error_counter(&self, labels: &ConnectorLabels) -> Counter {
self.errors.get_or_create(labels).clone()
@@ -475,6 +489,16 @@ impl Metrics {
.get()
}
+ #[cfg(test)]
+ pub(crate) fn get_sink_runs(&self, key: &str) -> u64 {
+ self.sink_runs
+ .get_or_create(&ConnectorLabels {
+ connector_key: key.to_owned(),
+ connector_type: ConnectorType::Sink,
+ })
+ .get()
+ }
+
pub fn get_errors(&self, key: &str, connector_type: ConnectorType) -> u64 {
self.errors
.get_or_create(&ConnectorLabels {
@@ -525,7 +549,7 @@ mod tests {
/// Every series the registry renders once each family holds a sample.
/// Counters carry exactly one `_total`, gauges and histogram buckets carry
/// the suffixes the OpenMetrics encoder gives them.
- const RENDERED_SERIES_NAMES: [&str; 13] = [
+ const RENDERED_SERIES_NAMES: [&str; 14] = [
"iggy_connectors_sources_total",
"iggy_connectors_sources_running",
"iggy_connectors_sinks_total",
@@ -536,6 +560,7 @@ mod tests {
"iggy_connector_messages_processed_total",
"iggy_connector_messages_filtered_total",
"iggy_connector_errors_total",
+ "iggy_connector_sink_runs_total",
"iggy_connector_stage_duration_seconds_sum",
"iggy_connector_stage_duration_seconds_count",
"iggy_connector_stage_duration_seconds_bucket",
@@ -556,6 +581,13 @@ mod tests {
metrics.increment_messages_processed("k", 1);
metrics.increment_messages_filtered("k", ConnectorType::Sink, 1);
metrics.increment_errors("k", ConnectorType::Sink);
+ metrics.inc_sink_runs_with_labels(
+ &ConnectorLabels {
+ connector_key: "k".to_owned(),
+ connector_type: ConnectorType::Sink,
+ },
+ 1,
+ );
metrics.observe_stage_duration(
"k",
ConnectorType::Sink,
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs
index 994cfc392..49459b803 100644
--- a/core/connectors/runtime/src/sink.rs
+++ b/core/connectors/runtime/src/sink.rs
@@ -353,11 +353,6 @@ pub(crate) async fn consume_messages(
let messages = std::mem::take(&mut batch);
let messages_count = messages.len();
metrics.inc_messages_consumed_with_labels(&labels.counter, messages_count as u64);
- let messages_metadata = MessagesMetadata {
- partition_id,
- current_offset,
- schema: decoder.schema(),
- };
if verbose {
info!(
"Processing {messages_count} messages for sink connector with ID: {}",
@@ -372,7 +367,8 @@ pub(crate) async fn consume_messages(
let start = Instant::now();
let result = process_messages(
plugin_id,
- messages_metadata,
+ partition_id,
+ current_offset,
&topic_metadata,
messages,
&consume,
@@ -386,7 +382,7 @@ pub(crate) async fn consume_messages(
// Total always records; sub-stages only on success (no 0-sample skew).
metrics.observe_stage_with_labels(&labels.stage_total, elapsed);
- let (processed_count, decode_us, prepare_us, ffi_us) = match &result {
+ let (processed_count, runs, decode_us, prepare_us, ffi_us) = match &result {
Ok(timing) => {
let prepare_elapsed = elapsed
.saturating_sub(timing.ffi_elapsed)
@@ -396,12 +392,13 @@ pub(crate) async fn consume_messages(
metrics.observe_stage_with_labels(&labels.stage_ffi, timing.ffi_elapsed);
(
timing.processed_count,
+ timing.runs,
benchmark::as_micros(timing.decode_elapsed),
benchmark::as_micros(prepare_elapsed),
benchmark::as_micros(timing.ffi_elapsed),
)
}
- Err(_) => (0, 0, 0, 0),
+ Err(_) => (0, 0, 0, 0, 0),
};
if benchmark {
@@ -413,6 +410,7 @@ pub(crate) async fn consume_messages(
current_offset,
messages_count,
processed_count,
+ runs,
decode_us,
prepare_us,
ffi_us,
@@ -549,7 +547,8 @@ pub(crate) async fn setup_sink_consumers(
#[allow(clippy::too_many_arguments)]
async fn process_messages(
plugin_id: u32,
- messages_metadata: MessagesMetadata,
+ partition_id: u32,
+ current_offset: u64,
topic_metadata: &TopicMetadata,
messages: Vec<IggyMessage>,
consume: &ConsumeCallback,
@@ -599,8 +598,17 @@ 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. No decoder mixes variants within one configured instance, but a
+ // transform can: `ProtoConvert` returns `Payload::Raw` when it can encode a
+ // message against its descriptor and `Payload::Proto` when it cannot, which
+ // turns on the message rather than the config. A uniform batch stays one run
+ // and one call.
+ let mut runs: Vec<(Schema, Vec<RawMessage>)> = Vec::with_capacity(1);
+ let mut remaining = decoded.len();
for message in decoded {
+ remaining -= 1;
let mut current_message = Some(message);
let mut transform_failed = false;
for transform in transforms.iter() {
@@ -674,6 +682,10 @@ async fn process_messages(
continue;
};
+ // Read the tag off the payload before `try_into_vec` consumes it. The
+ // decoder's own schema names the format it reads, not the variant it
+ // returned, and a transform may have changed the variant since.
+ let schema = message.payload.schema();
let Ok(payload) = message.payload.try_into_vec() else {
error!(
"Failed to get message payload for message with ID: {id}, offset: {offset} for sink connector with ID: {plugin_id}"
@@ -696,7 +708,7 @@ async fn process_messages(
None => vec![],
};
- messages.push(RawMessage {
+ let raw_message = RawMessage {
id,
offset,
checksum,
@@ -704,7 +716,23 @@ async fn process_messages(
origin_timestamp,
headers,
payload,
- });
+ };
+ match runs.last_mut() {
+ Some((run_schema, run)) if *run_schema == schema => run.push(raw_message),
+ _ => {
+ // Only the first run is sized to the batch. Reserving what is
+ // left for every run would cost O(n^2) slots on a batch that
+ // alternates variants, and any run after the first is rare
+ // enough to be worth growing on demand.
+ let mut run = if runs.is_empty() {
+ Vec::with_capacity(remaining + 1)
+ } else {
+ Vec::new()
+ };
+ run.push(raw_message);
+ runs.push((schema, run));
+ }
+ }
}
metrics.inc_errors_by_with_labels(&labels.counter, error_count);
@@ -712,7 +740,19 @@ async fn process_messages(
metrics.inc_messages_filtered_with_labels(&labels.counter, filtered_count);
}
- let processed_count = messages.len();
+ // A batch that lost every message still reaches the sink, as it always has.
+ // There is no payload to read a tag from, so it carries the stream's
+ // configured schema. That is the decoder's wire format rather than a payload
+ // variant, which is safe only because the run is empty: a tag is read back
+ // once per message, so an empty run never has it interpreted. Anything that
+ // starts branching on the tag for whole-batch behaviour has to revisit this.
+ if runs.is_empty() {
+ runs.push((decoder.schema(), Vec::new()));
+ }
+
+ // An empty batch is still one FFI call, so it is still one run.
+ let run_count = runs.len();
+ let mut processed_count = 0usize;
let topic_meta = postcard::to_allocvec(topic_metadata).map_err(|error| {
error!(
@@ -721,46 +761,59 @@ async fn process_messages(
RuntimeError::FailedToSerializeTopicMetadata
})?;
- 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 mut ffi_elapsed = Duration::ZERO;
+ for (schema, run) in runs {
+ let messages_metadata = MessagesMetadata {
+ partition_id,
+ current_offset,
+ 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(&RawMessages {
- schema: decoder.schema(),
- messages,
- })
- .map_err(|error| {
- error!("Failed to serialize messages for sink connector with ID: {plugin_id}. {error}");
- RuntimeError::FailedToSerializeRawMessages
- })?;
+ let run_len = run.len();
+ let messages = postcard::to_allocvec(&RawMessages {
+ schema,
+ messages: run,
+ })
+ .map_err(|error| {
+ error!("Failed to serialize messages for sink connector with ID: {plugin_id}. {error}");
+ RuntimeError::FailedToSerializeRawMessages
+ })?;
- let ffi_start = Instant::now();
- let result = (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 processed_count = if result == 0 {
- processed_count
- } else {
- error!(
- "Failed to consume {processed_count} messages for sink connector with ID: {plugin_id}, stream: {}, topic: {}, status: {result}",
- topic_metadata.stream, topic_metadata.topic
+ let ffi_start = Instant::now();
+ let result = (consume)(
+ plugin_id,
+ topic_meta.as_ptr(),
+ topic_meta.len(),
+ messages_meta.as_ptr(),
+ messages_meta.len(),
+ messages.as_ptr(),
+ messages.len(),
);
- metrics.inc_errors_with_labels(&labels.counter);
- 0
- };
+ ffi_elapsed += ffi_start.elapsed();
+ if result == 0 {
+ processed_count += run_len;
+ } else {
+ error!(
+ "Failed to consume {run_len} messages for sink connector with ID: {plugin_id}, stream: {}, topic: {}, schema: {schema}, status: {result}",
+ topic_metadata.stream, topic_metadata.topic
+ );
+ metrics.inc_errors_with_labels(&labels.counter);
+ }
+ }
+
+ // Counted once every call has been made, so a serialisation failure above
+ // cannot leave calls on the counter that never happened.
+ metrics.inc_sink_runs_with_labels(&labels.counter, run_count as u64);
Ok(SinkBatchTiming {
processed_count,
+ runs: run_count,
decode_elapsed,
ffi_elapsed,
})
@@ -768,6 +821,528 @@ async fn process_messages(
struct SinkBatchTiming {
processed_count: usize,
+ /// FFI calls the batch was split into, one per contiguous payload variant.
+ runs: usize,
decode_elapsed: Duration,
+ /// Summed over every run, so one batch is one `stage_ffi` sample however
+ /// many calls it took.
ffi_elapsed: Duration,
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use dashmap::DashMap;
+ use iggy::prelude::IggyMessageHeader;
+ use iggy_connector_sdk::transforms::TransformType;
+ use iggy_connector_sdk::transforms::{ProtoConvert, ProtoConvertConfig};
+ use iggy_connector_sdk::{Error, Payload};
+ use prost::Message as _;
+ use std::sync::LazyLock;
+ use std::sync::atomic::AtomicU32;
+
+ /// One entry per FFI call the stub sink received, keyed by plugin id so
+ /// tests sharing the binary do not read each other's batches.
+ static CONSUMED: LazyLock<DashMap<u32, Vec<ConsumedBatch>>> = LazyLock::new(DashMap::new);
+
+ static TEST_PLUGIN_ID: AtomicU32 = AtomicU32::new(u32::MAX / 2);
+
+ struct ConsumedBatch {
+ metadata_schema: Schema,
+ messages_schema: Schema,
+ payloads: Vec<Vec<u8>>,
+ offsets: Vec<u64>,
+ }
+
+ /// Runs the stub sink fails, keyed by plugin id and matched on the run's
+ /// schema, so a test reads the failure off the run split rather than off
+ /// a call counter.
+ static FAILING_SCHEMAS: LazyLock<DashMap<u32, Schema>> = LazyLock::new(DashMap::new);
+
+ /// High-water offset deliberately unrelated to any message offset in a
+ /// batch: `current_offset` is the partition's head from the poll, and a
+ /// helper that made it equal the last message's offset would let a test
+ /// assert the wrong contract without failing.
+ const TEST_CURRENT_OFFSET: u64 = 10_000;
+
+ /// Records the batch the FFI call carried and returns its run schema.
+ fn capture_batch(
+ plugin_id: u32,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> Schema {
+ let messages_meta =
+ unsafe { std::slice::from_raw_parts(messages_meta_ptr, messages_meta_len) };
+ let messages = unsafe { std::slice::from_raw_parts(messages_ptr, messages_len) };
+ let metadata = postcard::from_bytes::<MessagesMetadata>(messages_meta)
+ .expect("failed to deserialize messages metadata");
+ let raw = postcard::from_bytes::<RawMessages>(messages).expect("failed to deserialize");
+
+ CONSUMED.entry(plugin_id).or_default().push(ConsumedBatch {
+ metadata_schema: metadata.schema,
+ messages_schema: raw.schema,
+ offsets: raw.messages.iter().map(|message| message.offset).collect(),
+ payloads: raw
+ .messages
+ .into_iter()
+ .map(|message| message.payload)
+ .collect(),
+ });
+ metadata.schema
+ }
+
+ extern "C" fn capturing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ 0
+ }
+
+ /// Captures like `capturing_consume`, then fails the run whose schema the
+ /// test registered in `FAILING_SCHEMAS`.
+ extern "C" fn selectively_failing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ let schema = capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ match FAILING_SCHEMAS.get(&plugin_id) {
+ Some(failing) if *failing == schema => 1,
+ _ => 0,
+ }
+ }
+
+ extern "C" fn always_failing_consume(
+ plugin_id: u32,
+ _topic_meta_ptr: *const u8,
+ _topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ capture_batch(
+ plugin_id,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ );
+ 1
+ }
+
+ /// Rewrites every payload to the configured variant, standing in for a
+ /// third-party transform that changes the payload type.
+ struct RetaggingTransform {
+ payloads: Vec<Payload>,
+ next: std::sync::Mutex<usize>,
+ }
+
+ impl Transform for RetaggingTransform {
+ fn r#type(&self) -> TransformType {
+ TransformType::AvroConvert
+ }
+
+ fn transform(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ let mut next = self.next.lock().expect("transform counter was poisoned");
+ message.payload = self.payloads[*next % self.payloads.len()].clone();
+ *next += 1;
+ Ok(Some(message))
+ }
+ }
+
+ /// Drops every message, standing in for a filter transform that matches the
+ /// whole batch.
+ struct DroppingTransform;
+
+ impl Transform for DroppingTransform {
+ fn r#type(&self) -> TransformType {
+ TransformType::AvroConvert
+ }
+
+ fn transform(
+ &self,
+ _metadata: &TopicMetadata,
+ _message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ Ok(None)
+ }
+ }
+
+ fn next_plugin_id() -> u32 {
+ TEST_PLUGIN_ID.fetch_add(1, Ordering::Relaxed)
+ }
+
+ fn test_message(offset: u64, payload: Vec<u8>) -> IggyMessage {
+ IggyMessage {
+ header: IggyMessageHeader {
+ checksum: 0,
+ id: u128::from(offset) + 1,
+ offset,
+ timestamp: 0,
+ origin_timestamp: 0,
+ user_headers_length: 0,
+ payload_length: payload.len() as u32,
+ reserved: 0,
+ },
+ payload: payload.into(),
+ user_headers: None,
+ }
+ }
+
+ fn avro_schema() -> apache_avro::Schema {
+ apache_avro::Schema::parse_str(
+ r#"{"type":"record","name":"Event","fields":[{"name":"id","type":"long"}]}"#,
+ )
+ .expect("failed to parse Avro schema")
+ }
+
+ fn avro_datum(schema: &apache_avro::Schema, id: i64) -> Vec<u8> {
+ let record = apache_avro::types::Value::Record(vec![(
+ "id".to_owned(),
+ apache_avro::types::Value::Long(id),
+ )]);
+ apache_avro::writer::datum::GenericDatumWriter::builder(schema)
+ .build()
+ .expect("failed to build Avro writer")
+ .write_value_to_vec(record)
+ .expect("failed to encode Avro datum")
+ }
+
+ async fn run(
+ plugin_id: u32,
+ decoder: Arc<dyn StreamDecoder>,
+ transforms: Vec<Arc<dyn Transform>>,
+ messages: Vec<IggyMessage>,
+ ) -> SinkBatchTiming {
+ run_with(plugin_id, capturing_consume, decoder, transforms, messages)
+ .await
+ .0
+ }
+
+ /// Drives one batch through `process_messages` with the given stub sink
+ /// and hands back the metrics it wrote, so a test can read counters.
+ async fn run_with(
+ plugin_id: u32,
+ consume: ConsumeCallback,
+ decoder: Arc<dyn StreamDecoder>,
+ transforms: Vec<Arc<dyn Transform>>,
+ messages: Vec<IggyMessage>,
+ ) -> (SinkBatchTiming, Arc<Metrics>) {
+ let metrics = Arc::new(Metrics::init());
+ let labels = SinkLabels::new("test_sink");
+ let topic_metadata = TopicMetadata {
+ stream: "test_stream".to_owned(),
+ topic: "test_topic".to_owned(),
+ };
+
+ let timing = process_messages(
+ plugin_id,
+ 0,
+ TEST_CURRENT_OFFSET,
+ &topic_metadata,
+ messages,
+ &consume,
+ &transforms,
+ &decoder,
+ &metrics,
+ &labels,
+ )
+ .await
+ .expect("processing the batch should succeed");
+ (timing, metrics)
+ }
+
+ /// Four messages retagged `Text, Text, Raw, Text`, which the runtime
+ /// splits into three runs.
+ fn split_batch() -> (Arc<RetaggingTransform>, Vec<IggyMessage>) {
+ let transform = Arc::new(RetaggingTransform {
+ payloads: vec![
+ Payload::Text("first".to_owned()),
+ Payload::Text("second".to_owned()),
+ Payload::Raw(vec![9]),
+ Payload::Text("fourth".to_owned()),
+ ],
+ next: std::sync::Mutex::new(0),
+ });
+ let messages = (0..4)
+ .map(|offset| test_message(offset, br#"{"id":1}"#.to_vec()))
+ .collect();
+ (transform, messages)
+ }
+
+ fn captured(plugin_id: u32) -> Vec<ConsumedBatch> {
+ CONSUMED
+ .remove(&plugin_id)
+ .map(|(_, batches)| batches)
+ .unwrap_or_default()
+ }
+
+ #[tokio::test]
+ async fn given_an_avro_stream_when_batch_is_tagged_should_use_the_decoded_payload_schema() {
+ let plugin_id = next_plugin_id();
+ let schema = avro_schema();
+ let decoder = Arc::new(
+ AvroStreamDecoder::try_new(AvroConfig {
+ schema_json: Some(schema.canonical_form()),
+ ..AvroConfig::default()
+ })
+ .expect("failed to build the Avro decoder"),
+ );
+ let messages = vec![
+ test_message(0, avro_datum(&schema, 1)),
+ test_message(1, avro_datum(&schema, 2)),
+ ];
+
+ let timing = run(plugin_id, decoder, Vec::new(), messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 2);
+ assert_eq!(batches.len(), 1, "a uniform batch is one FFI call");
+ // The decoder reads Avro but returns JSON, so JSON is what the sink is
+ // told it has.
+ assert_eq!(batches[0].metadata_schema, Schema::Json);
+ assert_eq!(batches[0].messages_schema, Schema::Json);
+ for payload in &batches[0].payloads {
+ serde_json::from_slice::<serde_json::Value>(payload).expect("payload should be JSON");
+ }
+ }
+
+ #[tokio::test]
+ async fn given_a_transform_changing_the_variant_when_batch_is_tagged_should_follow_the_transform()
+ {
+ let plugin_id = next_plugin_id();
+ let transform = Arc::new(RetaggingTransform {
+ payloads: vec![Payload::Avro(vec![1, 2, 3])],
+ next: std::sync::Mutex::new(0),
+ });
+ let messages = vec![test_message(0, br#"{"id":1}"#.to_vec())];
+
+ run(plugin_id, Schema::Json.decoder(), vec![transform], messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].metadata_schema, Schema::Avro);
+ assert_eq!(batches[0].messages_schema, Schema::Avro);
+ }
+
+ #[tokio::test]
+ async fn given_mixed_payload_variants_when_batch_is_processed_should_split_into_runs() {
+ let plugin_id = next_plugin_id();
+ let (transform, messages) = split_batch();
+
+ let timing = run(plugin_id, Schema::Json.decoder(), vec![transform], messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 4, "no message may be lost");
+ assert_eq!(batches.len(), 3);
+ assert_eq!(batches[0].metadata_schema, Schema::Text);
+ assert_eq!(batches[0].offsets, vec![0, 1]);
+ assert_eq!(batches[1].metadata_schema, Schema::Raw);
+ assert_eq!(batches[1].offsets, vec![2]);
+ assert_eq!(batches[2].metadata_schema, Schema::Text);
+ assert_eq!(batches[2].offsets, vec![3]);
+ }
+
+ #[tokio::test]
+ async fn given_a_failing_run_when_a_batch_is_split_should_debit_only_that_run() {
+ let plugin_id = next_plugin_id();
+ FAILING_SCHEMAS.insert(plugin_id, Schema::Raw);
+ let (transform, messages) = split_batch();
+
+ let (timing, metrics) = run_with(
+ plugin_id,
+ selectively_failing_consume,
+ Schema::Json.decoder(),
+ vec![transform],
+ messages,
+ )
+ .await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 3, "only the failing run is lost");
+ assert_eq!(timing.runs, 3);
+ assert_eq!(batches.len(), 3, "a failing run does not stop later runs");
+ assert_eq!(batches[2].offsets, vec![3]);
+ assert_eq!(metrics.get_errors("test_sink", ConnectorType::Sink), 1);
+ assert_eq!(metrics.get_sink_runs("test_sink"), 3);
+ }
+
+ #[tokio::test]
+ async fn given_every_run_failing_when_a_batch_is_split_should_report_no_processed_messages() {
+ let plugin_id = next_plugin_id();
+ let (transform, messages) = split_batch();
+
+ let (timing, metrics) = run_with(
+ plugin_id,
+ always_failing_consume,
+ Schema::Json.decoder(),
+ vec![transform],
+ messages,
+ )
+ .await;
+
+ assert_eq!(timing.processed_count, 0);
+ assert_eq!(captured(plugin_id).len(), 3, "every run is still attempted");
+ assert_eq!(
+ metrics.get_errors("test_sink", ConnectorType::Sink),
+ 3,
+ "one error per failed run"
+ );
+ }
+
+ #[tokio::test]
+ async fn given_a_batch_that_lost_every_message_when_processed_should_still_be_one_run() {
+ let plugin_id = next_plugin_id();
+ let messages = (0..3)
+ .map(|offset| test_message(offset, br#"{"id":1}"#.to_vec()))
+ .collect();
+
+ let timing = run(
+ plugin_id,
+ Schema::Json.decoder(),
+ vec![Arc::new(DroppingTransform)],
+ messages,
+ )
+ .await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 0);
+ assert_eq!(timing.runs, 1, "an empty batch is still one FFI call");
+ assert_eq!(batches.len(), 1);
+ assert!(batches[0].offsets.is_empty());
+ // Nothing is left to read a variant from, so the run carries the
+ // stream's configured schema rather than a payload's.
+ assert_eq!(batches[0].metadata_schema, Schema::Json);
+ }
+
+ #[tokio::test]
+ async fn given_a_uniform_batch_when_batch_is_processed_should_count_one_run() {
+ let plugin_id = next_plugin_id();
+ let messages = (0..3)
+ .map(|offset| test_message(offset, br#"{"id":1}"#.to_vec()))
+ .collect();
+
+ let (timing, metrics) = run_with(
+ plugin_id,
+ capturing_consume,
+ Schema::Json.decoder(),
+ Vec::new(),
+ messages,
+ )
+ .await;
+
+ assert_eq!(timing.runs, 1);
+ assert_eq!(metrics.get_sink_runs("test_sink"), 1);
+ assert_eq!(captured(plugin_id).len(), 1);
+ }
+
+ #[tokio::test]
+ async fn given_no_surviving_messages_when_batch_is_processed_should_still_call_the_sink_once() {
+ let plugin_id = next_plugin_id();
+ // An Avro decoder, so the fallback tag cannot be confused with
+ // `Schema::default()`.
+ let decoder = Arc::new(
+ AvroStreamDecoder::try_new(AvroConfig {
+ schema_json: Some(avro_schema().canonical_form()),
+ ..AvroConfig::default()
+ })
+ .expect("failed to build the Avro decoder"),
+ );
+ let messages = vec![test_message(0, b"not an avro datum".to_vec())];
+
+ let timing = run(plugin_id, decoder, Vec::new(), messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 0);
+ assert_eq!(batches.len(), 1);
+ assert!(batches[0].offsets.is_empty());
+ // Nothing survived to read a tag from, so the stream's configured
+ // schema stands in.
+ assert_eq!(batches[0].metadata_schema, Schema::Avro);
+ }
+
+ #[tokio::test]
+ async fn given_a_proto_convert_transform_when_a_batch_mixes_variants_should_split_into_runs() {
+ // `ProtoConvert` encodes a top-level object against its descriptor and
+ // falls back to proto text for anything else, so one configured
+ // instance returns `Raw` for one message and `Proto` for the next.
+ let plugin_id = next_plugin_id();
+ let descriptor = protox_parse::parse(
+ "record.proto",
+ r#"syntax = "proto3";
+ message StringRecord {
+ string first = 1;
+ }"#,
+ )
+ .expect("test schema must parse");
+ let transform = Arc::new(ProtoConvert::new(ProtoConvertConfig {
+ source_format: Schema::Json,
+ target_format: Schema::Proto,
+ message_type: Some("StringRecord".to_owned()),
+ descriptor_set: Some(
+ prost_types::FileDescriptorSet {
+ file: vec![descriptor],
+ }
+ .encode_to_vec(),
+ ),
+ ..ProtoConvertConfig::default()
+ }));
+ let messages = vec![
+ test_message(0, br#"{"first":"encoded"}"#.to_vec()),
+ test_message(1, br#"[1,2,3]"#.to_vec()),
+ test_message(2, br#"{"first":"encoded"}"#.to_vec()),
+ ];
+
+ let timing = run(plugin_id, Schema::Json.decoder(), vec![transform], messages).await;
+ let batches = captured(plugin_id);
+
+ assert_eq!(timing.processed_count, 3, "no message may be lost");
+ assert_eq!(
+ batches.len(),
+ 3,
+ "each variant change starts a new FFI call"
+ );
+ assert_eq!(batches[0].metadata_schema, Schema::Raw);
+ assert_eq!(batches[0].offsets, vec![0]);
+ assert_eq!(batches[1].metadata_schema, Schema::Proto);
+ assert_eq!(batches[1].offsets, vec![1]);
+ assert_eq!(batches[2].metadata_schema, Schema::Raw);
+ assert_eq!(batches[2].offsets, vec![2]);
+
+ // The tag has to survive the trip back, or the sink sees a variant the
+ // transform never produced.
+ let rebuilt =
+ Payload::try_from_schema(batches[1].messages_schema, batches[1].payloads[0].clone())
+ .expect("the proto run must rebuild");
+ assert_eq!(rebuilt.schema(), Schema::Proto);
+ }
+}
diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml
index c96cee0dd..a114427db 100644
--- a/core/connectors/sdk/Cargo.toml
+++ b/core/connectors/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_sdk"
-version = "0.4.0"
+version = "0.5.0"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sdk/README.md b/core/connectors/sdk/README.md
index d91628058..f058291d4 100644
--- a/core/connectors/sdk/README.md
+++ b/core/connectors/sdk/README.md
@@ -54,7 +54,7 @@ value.static = "hello"
`retry_async` runs an operation that fails with `Err` and retries it while `should_retry` accepts the error. It owns attempt counting, backoff and the per-retry log, and returns `RetryFailure { error, attempts, exhausted }` so the caller logs the terminal failure. `retry_backoff` computes a single delay for a loop that cannot use `retry_async`, such as `HttpRetryMiddleware`, which retries on an `Ok` response rather than an `Err`. Its `retry` argument is 1-based.
-Two items changed in a way that breaks out-of-tree plugins, so those plugins must be rebuilt against the current source:
+Two symbols were removed and one field changed meaning, each in a way that breaks out-of-tree plugins, so those plugins must be rebuilt against the current source:
| Removed | Replacement |
| --- | --- |
@@ -63,6 +63,8 @@ Two items changed in a way that breaks out-of-tree plugins, so those plugins mus
Both types carry `(u32, Duration, Duration)` and the two delay roles cross over, so a field-by-field rename compiles and swaps the base delay for the cap. Map the fields by name.
+`MessagesMetadata.schema` and `RawMessages.schema` now name the variant the `Payload` holds rather than the wire format the stream's decoder reads. `Schema::Proto` means a `Payload::Proto` string on the sink path where it meant protobuf wire bytes before, so a plugin built against 0.4.0 rebuilds the run as the wrong variant with no error anywhere.
+
## Protocol Buffers Support
The SDK includes support for Protocol Buffers (protobuf) format with both encoding and decoding capabilities. Protocol Buffers provide efficient serialization and are particularly useful for high-performance data streaming scenarios.
@@ -156,12 +158,16 @@ Create the stream and topic, then start the runtime from the repository root:
IGGY_CONNECTORS_CONFIG_PATH=connectors.toml ./target/release/iggy-connectors
```
-The source sends 100 records, then continues polling without new messages. Stdout logs message offsets and the serialized JSON envelope bytes, containing `type_url` and base64 `value`. The sink's `raw` schema also determines how the plugin receives those transformed bytes.
+The source sends 100 records, then continues polling without new messages. Stdout logs message offsets and the serialized JSON envelope bytes, containing `type_url` and base64 `value`. The batch handed to the plugin is tagged with the payload's own schema rather than the stream's configured `raw`: `proto_convert` leaves a `Payload::Json`, so the plugin receives a `json` batch.
The format-conversion transforms define no per-key defaults. Every non-optional key shown above must be present, or the configuration fails to deserialize (`schema_path`, `message_type`, `field_mappings`, and `descriptor_set` are optional).
The two `[[streams]]` shapes differ: a source produces to a single `topic` and can tune batching via `batch_length` and `linger_time`, while a sink consumes from a list of `topics` and can additionally set `batch_length`, `poll_interval`, and `consumer_group`.
+Transforms are keyed by type, so one connector configures at most one `proto_convert`.
+
+The order of a transform chain is not defined. The runtime builds the chain from a map keyed by transform type, so two transforms on one connector can run in either order from one process to the next. Do not configure a chain whose result depends on which transform runs first.
+
### Key Configuration Options
#### Programmatic Encoder and Decoder Configuration
@@ -175,7 +181,7 @@ These are SDK configuration fields, not Random or Stdout `plugin_config` keys. T
#### Transform Options
- **`proto_convert`**: Transform for converting between protobuf and other formats
-- **`source_format`** / **`target_format`**: Formats to convert between - any schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`)
+- **`source_format`** / **`target_format`**: Formats to convert between - any schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`). Only `flatbuffer_convert` checks `source_format` against the payload it was handed and rejects a mismatch. `proto_convert` and `avro_convert` dispatch on the format pair with no up-front guard, which is why the `schema = "raw"` plus `source_format = "proto"` example above works
- **`preserve_unknown_fields`**: Accepted by `proto_convert`, but currently has no effect
- **`include_paths`**: Additional directories searched for imported `.proto` files
- **`field_mappings`**: Renames fields in a JSON input object before conversion (e.g., `"old_field" = "new_field"`)
@@ -352,6 +358,7 @@ fn main() -> Result<(), Error> {
- **Manual Loading**: `load_schema()` reloads the configured source. Missing or unreadable files, invalid protobuf syntax, compilation failures, and malformed descriptor bytes return errors and preserve an already-loaded schema. `update_config(config, true)` also restores the previous configuration on error; `false` changes the configuration while retaining the cached schema.
- **Fallbacks**: Absent schema sources or an unmatched `message_type` can return `Ok(())` without an active message descriptor. Successful reloads into fallback mode clear the previous descriptor. Check the actual encoded/decoded result when validating a schema setup.
- **Encoding Errors**: Errors encoding a loaded message descriptor are returned to the caller. The encoder does not retry that message as Any. The decoder attempts Any after a schema decoding error.
-- **Transform Configuration**: Create a new converter to change its configuration. `load_schema()` can reload its existing source. Without a descriptor, JSON-to-protobuf conversion produces JSON text in `Payload::Proto`, not a schema-encoded binary message.
+- **Transform Configuration**: Create a new converter to change its configuration. `load_schema()` can reload its existing source. Without a descriptor, JSON-to-protobuf conversion produces JSON text in `Payload::Proto`, not a schema-encoded binary message. With a descriptor it encodes a top-level JSON object to `Payload::Raw` and falls back to `Payload::Proto` for anything else, so one configured instance can return either variant depending on the message.
+- **Two Schema Inverses**: `Schema::Proto` means protobuf wire bytes when a source plugin sets `ProducedMessages::schema`, and a `Payload::Proto` string when the runtime tags a sink batch from `Payload::schema()`. `Schema::try_into_payload` reads the first and `Payload::try_from_schema` the second. Sink plugins use `Payload::try_from_schema`, which round-trips every variant; the source path keeps the Any-decoding arm.
- **Format Options**: Encoder `preserve_unknown_fields`, `compact_encoding`, `validate_message`, and `deterministic_encoding` are accepted but have no effect. Decoder `preserve_unknown_fields` retains unknown varints as numbers and length-delimited data as base64; fixed-width unknown fields become placeholders. It does not retain the original wire encoding. See the [Transforms page](https://iggy.apache.org/docs/connectors/transforms) for conversion-specific limits.
- Protocol Buffers provide efficient binary serialization compared to JSON
diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs
index 0d6efc51b..1fe5a67e4 100644
--- a/core/connectors/sdk/src/lib.rs
+++ b/core/connectors/sdk/src/lib.rs
@@ -28,6 +28,7 @@ use encoders::{
use iggy::prelude::{HeaderKey, HeaderValue};
use prost::Message;
use serde::{Deserialize, Serialize};
+use std::borrow::Cow;
use std::collections::BTreeMap;
use std::sync::{Arc, OnceLock};
use strum_macros::{Display, IntoStaticStr};
@@ -134,7 +135,13 @@ pub trait Sink: Send + Sync {
/// Invoked when the sink is initialized, allowing it to perform any necessary setup.
async fn open(&mut self) -> Result<(), Error>;
- /// Invoked every time a batch of messages is received from the configured stream(s) and topic(s).
+ /// Invoked for each run of messages polled from the configured stream(s) and topic(s).
+ ///
+ /// One poll can reach the plugin as more than one call. The runtime groups a batch into
+ /// contiguous runs of a single payload variant and sends each run on its own, so a batch
+ /// whose variant changes partway through arrives as several calls, every one of them
+ /// repeating the same `messages_metadata.current_offset`. A batch that lost every message
+ /// to the decoder or to a transform still arrives, as one call carrying no messages.
async fn consume(
&self,
topic_metadata: &TopicMetadata,
@@ -157,6 +164,87 @@ 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,
+ Payload::FlatBuffer(_) => Schema::FlatBuffer,
+ Payload::Avro(_) => Schema::Avro,
+ }
+ }
+
+ /// Rebuilds a payload from a tag produced by `Payload::schema`.
+ ///
+ /// The variant-preserving inverse of `Payload::schema`, and not the same
+ /// thing as `Schema::try_into_payload`, which reads a tag naming the wire
+ /// format a source plugin sent. The two disagree on `Schema::Proto`: it
+ /// means protobuf wire bytes to a source and a `Payload::Proto` string
+ /// here, so the sink path needs its own inverse to get its variant back.
+ pub fn try_from_schema(schema: Schema, mut value: Vec<u8>) -> Result<Self, Error> {
+ match schema {
+ Schema::Json => Ok(Payload::Json(
+ simd_json::to_owned_value(&mut value).map_err(|_| Error::InvalidJsonPayload)?,
+ )),
+ Schema::Raw => Ok(Payload::Raw(value)),
+ Schema::Text => Ok(Payload::Text(
+ String::from_utf8(value).map_err(|_| Error::InvalidTextPayload)?,
+ )),
+ Schema::Proto => Ok(Payload::Proto(
+ String::from_utf8(value).map_err(|_| Error::InvalidProtobufPayload)?,
+ )),
+ Schema::FlatBuffer => Ok(Payload::FlatBuffer(value)),
+ Schema::Avro => Ok(Payload::Avro(value)),
+ }
+ }
+
+ /// The JSON document this payload holds, when it holds one.
+ ///
+ /// A `Payload::Json` is borrowed as it is. A `Payload::Proto` is parsed,
+ /// because `proto_convert` puts the JSON text it was handed there whenever
+ /// it has no descriptor or the encode fails, and puts arbitrary text there
+ /// on its text and raw paths. The parse can therefore fail by design, and
+ /// `None` means the text has to be treated as text. Every other variant is
+ /// `None`.
+ ///
+ /// Proto text is parsed from a copy: simd_json mutates its buffer even when
+ /// the parse fails, and the text has to survive for the fallback.
+ pub fn json_document(&self) -> Option<Cow<'_, simd_json::OwnedValue>> {
+ match self {
+ Payload::Json(value) => Some(Cow::Borrowed(value)),
+ Payload::Proto(text) => {
+ let mut bytes = text.as_bytes().to_vec();
+ simd_json::to_owned_value(&mut bytes).ok().map(Cow::Owned)
+ }
+ _ => None,
+ }
+ }
+
+ /// The same payload, with proto text that holds a JSON document replaced by
+ /// that document.
+ ///
+ /// The consuming counterpart to `json_document`, for a sink that wants the
+ /// descriptor-less `proto_convert` fallback to take its existing JSON path.
+ /// A `Payload::Json` already holds the document and is returned as it is,
+ /// and proto text that is not JSON stays `Payload::Proto` for the sink's
+ /// text handling. Every other variant is returned unchanged.
+ pub fn into_json_document(self) -> Self {
+ match self.json_document() {
+ Some(Cow::Owned(document)) => Payload::Json(document),
+ _ => self,
+ }
+ }
+
/// Consuming conversion — transfers ownership of inner buffers.
pub fn try_into_vec(self) -> Result<Vec<u8>, Error> {
match self {
@@ -240,6 +328,13 @@ pub enum Schema {
}
impl Schema {
+ /// Rebuilds a payload from a tag naming the wire format its bytes are in.
+ ///
+ /// Carries the source path, where a plugin sets `ProducedMessages::schema`
+ /// to describe the bytes it produced. `Schema::Proto` therefore means
+ /// protobuf wire bytes and is read as a `prost_types::Any`. Use
+ /// `Payload::try_from_schema` to invert a tag that came from
+ /// `Payload::schema` instead.
pub fn try_into_payload(self, mut value: Vec<u8>) -> Result<Payload, Error> {
match self {
Schema::Json => Ok(Payload::Json(
@@ -294,11 +389,20 @@ pub struct TopicMetadata {
pub topic: String,
}
+/// Describes the run of messages carried by one `Sink::consume` call.
#[repr(C)]
#[derive(Debug, Serialize, Deserialize)]
pub struct MessagesMetadata {
+ /// The partition the run was polled from.
pub partition_id: u32,
+ /// The partition's high-water offset at the time of the poll, not the offset of the last
+ /// message in this run. Every call the poll produces repeats it, so a sink that keys a
+ /// commit, a file name or a table version on it has to tolerate the repeat.
pub current_offset: u64,
+ /// The variant every `Payload` in this run holds, which is not always the stream's
+ /// configured `schema`: a decoder may return a different form than the wire format it
+ /// reads, and a transform may change the variant again. An empty run has no payload to
+ /// read a variant from and carries the stream's configured schema instead.
pub schema: Schema,
}
@@ -471,3 +575,231 @@ pub enum Error {
#[error("State provider latched after a permanent state error")]
StateLatched,
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn all_payloads() -> Vec<(Payload, Schema)> {
+ vec![
+ (Payload::Json(simd_json::json!({"id": 1})), Schema::Json),
+ (Payload::Json(simd_json::json!([1, 2, 3])), Schema::Json),
+ (Payload::Json(simd_json::json!("scalar")), Schema::Json),
+ (Payload::Json(simd_json::json!(null)), Schema::Json),
+ (Payload::Raw(vec![1, 2, 3]), Schema::Raw),
+ (Payload::Raw(Vec::new()), Schema::Raw),
+ (Payload::Text("hello".to_owned()), Schema::Text),
+ (Payload::Text(String::new()), Schema::Text),
+ (Payload::Proto("proto text".to_owned()), Schema::Proto),
+ (Payload::Proto(String::new()), Schema::Proto),
+ (Payload::FlatBuffer(vec![4, 5, 6]), Schema::FlatBuffer),
+ (Payload::FlatBuffer(Vec::new()), Schema::FlatBuffer),
+ (Payload::Avro(vec![7, 8, 9]), Schema::Avro),
+ (Payload::Avro(Vec::new()), Schema::Avro),
+ ]
+ }
+
+ #[test]
+ fn given_a_json_payload_when_the_document_is_read_should_borrow_it() {
+ let payload = Payload::Json(simd_json::json!({"id": 1}));
+
+ let document = payload
+ .json_document()
+ .expect("a JSON payload is a document");
+
+ assert!(matches!(document, Cow::Borrowed(_)));
+ assert_eq!(*document, simd_json::json!({"id": 1}));
+ }
+
+ #[test]
+ fn given_proto_text_holding_json_when_the_document_is_read_should_parse_it() {
+ let payload = Payload::Proto(r#"{"id": 1, "name": "row-1"}"#.to_owned());
+
+ let document = payload
+ .json_document()
+ .expect("proto text holding JSON is a document");
+
+ assert!(matches!(document, Cow::Owned(_)));
+ assert_eq!(*document, simd_json::json!({"id": 1, "name": "row-1"}));
+ }
+
+ #[test]
+ fn given_proto_text_that_is_not_json_when_the_document_is_read_should_leave_the_text_intact() {
+ let text = r#"binary_data: "AQID""#;
+ let payload = Payload::Proto(text.to_owned());
+
+ assert!(payload.json_document().is_none());
+ // simd_json overwrites its input buffer on a failed parse, so this
+ // only holds because the parse ran on a copy.
+ let Payload::Proto(kept) = &payload else {
+ panic!("the variant must not change");
+ };
+ assert_eq!(kept, text);
+ }
+
+ #[test]
+ fn given_a_payload_that_is_not_json_or_proto_when_the_document_is_read_should_return_none() {
+ for payload in [
+ Payload::Text(r#"{"id": 1}"#.to_owned()),
+ Payload::Raw(br#"{"id": 1}"#.to_vec()),
+ Payload::FlatBuffer(vec![1, 2, 3]),
+ Payload::Avro(vec![1, 2, 3]),
+ ] {
+ assert!(
+ payload.json_document().is_none(),
+ "{payload} must not be read as a document"
+ );
+ }
+ }
+
+ #[test]
+ fn given_proto_text_holding_json_when_converted_should_become_a_json_payload() {
+ let payload = Payload::Proto(r#"{"id": 1, "name": "row-1"}"#.to_owned());
+
+ let Payload::Json(document) = payload.into_json_document() else {
+ panic!("proto text holding JSON becomes a JSON payload");
+ };
+
+ assert_eq!(document, simd_json::json!({"id": 1, "name": "row-1"}));
+ }
+
+ #[test]
+ fn given_proto_text_that_is_not_json_when_converted_should_keep_the_text_intact() {
+ let text = r#"binary_data: "AQID""#;
+
+ let converted = Payload::Proto(text.to_owned()).into_json_document();
+
+ // The text survives only because the parse ran on a copy; an in-place
+ // parse would leave it overwritten here.
+ let Payload::Proto(kept) = &converted else {
+ panic!("the variant must not change");
+ };
+ assert_eq!(kept, text);
+ }
+
+ #[test]
+ fn given_a_json_payload_when_converted_should_return_it_unchanged() {
+ let payload = Payload::Json(simd_json::json!({"id": 1}));
+
+ let Payload::Json(document) = payload.into_json_document() else {
+ panic!("a JSON payload stays a JSON payload");
+ };
+
+ assert_eq!(document, simd_json::json!({"id": 1}));
+ }
+
+ #[test]
+ fn given_a_payload_that_is_not_json_or_proto_when_converted_should_return_it_unchanged() {
+ for payload in [
+ Payload::Text(r#"{"id": 1}"#.to_owned()),
+ Payload::Raw(br#"{"id": 1}"#.to_vec()),
+ Payload::FlatBuffer(vec![1, 2, 3]),
+ Payload::Avro(vec![1, 2, 3]),
+ ] {
+ let expected_schema = payload.schema();
+ let expected_bytes = payload
+ .clone()
+ .try_into_vec()
+ .expect("the payload should serialise");
+
+ let converted = payload.into_json_document();
+
+ assert_eq!(converted.schema(), expected_schema);
+ assert_eq!(
+ converted
+ .try_into_vec()
+ .expect("the payload should serialise"),
+ expected_bytes,
+ "the bytes must survive as well as the variant"
+ );
+ }
+ }
+
+ #[test]
+ fn given_every_payload_variant_when_schema_is_read_should_name_that_variant() {
+ for (payload, expected) in all_payloads() {
+ assert_eq!(
+ payload.schema(),
+ expected,
+ "wrong schema for {payload}, expected {expected}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_a_payload_when_round_tripped_through_its_own_schema_should_keep_the_variant_and_the_bytes()
+ {
+ // The JSON rows are small objects, arrays and scalars, so their
+ // serialised bytes are stable. A row with many keys would need a value
+ // comparison instead, because simd_json objects do not keep key order
+ // past their small-map threshold.
+ for (payload, schema) in all_payloads() {
+ let bytes = payload
+ .try_to_bytes()
+ .unwrap_or_else(|error| panic!("failed to serialize {schema} payload: {error}"));
+ let rebuilt = Payload::try_from_schema(schema, bytes.clone())
+ .unwrap_or_else(|error| panic!("failed to rebuild {schema} payload: {error}"));
+
+ assert_eq!(
+ rebuilt.schema(),
+ schema,
+ "round trip through {schema} produced {rebuilt}"
+ );
+ assert_eq!(
+ rebuilt
+ .try_to_bytes()
+ .unwrap_or_else(|error| panic!("failed to serialize {rebuilt}: {error}")),
+ bytes,
+ "round trip through {schema} changed the payload bytes"
+ );
+ }
+ }
+
+ #[test]
+ fn given_proto_text_when_rebuilt_from_a_variant_tag_should_return_the_proto_payload() {
+ let payload = Payload::Proto(r#"{"id":1}"#.to_owned());
+ let bytes = payload.try_into_vec().expect("failed to serialize");
+ let rebuilt = Payload::try_from_schema(Schema::Proto, bytes).expect("failed to rebuild");
+
+ let Payload::Proto(text) = rebuilt else {
+ panic!("expected a proto payload, got {rebuilt}");
+ };
+ assert_eq!(text, r#"{"id":1}"#);
+ }
+
+ #[test]
+ fn given_non_utf8_bytes_when_rebuilt_as_proto_from_a_variant_tag_should_fail() {
+ let error = Payload::try_from_schema(Schema::Proto, vec![0xff, 0xfe])
+ .expect_err("non UTF-8 bytes cannot be a proto text payload");
+
+ assert_eq!(error, Error::InvalidProtobufPayload);
+ }
+
+ #[test]
+ fn given_protobuf_wire_bytes_when_rebuilt_from_a_wire_tag_should_stay_an_any_document() {
+ // The source path keeps its own inverse: a plugin tagging
+ // `Schema::Proto` really does send protobuf, so those bytes are read as
+ // a `prost_types::Any` rather than as text.
+ let any = prost_types::Any {
+ type_url: "type.googleapis.com/test.Event".to_owned(),
+ value: vec![1, 2, 3],
+ };
+ let rebuilt = Schema::Proto
+ .try_into_payload(any.encode_to_vec())
+ .expect("failed to rebuild");
+
+ assert_eq!(rebuilt.schema(), Schema::Json);
+ }
+
+ #[test]
+ fn given_bytes_that_are_not_an_any_when_rebuilt_from_a_wire_tag_should_fall_back_to_raw() {
+ // The other half of the source arm. A plugin can tag `Schema::Proto`
+ // and send something that is not an `Any`, and those bytes still have
+ // to reach the sink rather than being dropped.
+ let rebuilt = Schema::Proto
+ .try_into_payload(b"not protobuf wire bytes".to_vec())
+ .expect("failed to rebuild");
+
+ assert_eq!(rebuilt.schema(), Schema::Raw);
+ }
+}
diff --git a/core/connectors/sdk/src/sink.rs b/core/connectors/sdk/src/sink.rs
index cbf8974f8..9891d2243 100644
--- a/core/connectors/sdk/src/sink.rs
+++ b/core/connectors/sdk/src/sink.rs
@@ -21,7 +21,9 @@ use tracing::{error, info};
use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt};
use crate::log::{CallbackLayer, LogCallback};
-use crate::{ConsumedMessage, MessagesMetadata, RawMessages, Sink, TopicMetadata, get_runtime};
+use crate::{
+ ConsumedMessage, MessagesMetadata, Payload, RawMessages, Sink, TopicMetadata, get_runtime,
+};
pub type ConsumeCallback = extern "C" fn(
plugin_id: u32,
@@ -196,7 +198,12 @@ impl<T: Sink + std::fmt::Debug> SinkContainer<T> {
}
};
- let payload = match messages_metadata.schema.try_into_payload(message.payload) {
+ // The runtime tags each run from `Payload::schema`, so the
+ // tag names a variant here rather than a wire format.
+ let payload = match Payload::try_from_schema(
+ messages_metadata.schema,
+ message.payload,
+ ) {
Ok(payload) => payload,
Err(err) => {
error!(
@@ -323,3 +330,141 @@ macro_rules! sink_connector {
}
};
}
+
+#[cfg(test)]
+mod tests {
+ use crate::sink::SinkContainer;
+ use crate::{
+ ConsumedMessage, Error, MessagesMetadata, Payload, RawMessage, RawMessages, Schema, Sink,
+ TopicMetadata,
+ };
+ use async_trait::async_trait;
+ use std::sync::{Arc, Mutex};
+
+ /// Keeps the payloads the container rebuilt, which is the only observable
+ /// the FFI entry point produces.
+ #[derive(Debug, Default)]
+ struct RecordingSink {
+ consumed: Arc<Mutex<Vec<Payload>>>,
+ }
+
+ #[async_trait]
+ impl Sink for RecordingSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ _topic_metadata: &TopicMetadata,
+ _messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ let mut consumed = self.consumed.lock().expect("recorder was poisoned");
+ consumed.extend(messages.into_iter().map(|message| message.payload));
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ Ok(())
+ }
+ }
+
+ /// Drives one run through the container the way the runtime does: postcard
+ /// bytes in, and the payloads the sink was handed out.
+ fn consume_one(schema: Schema, payload: Vec<u8>) -> Vec<Payload> {
+ let consumed = Arc::new(Mutex::new(Vec::new()));
+ let container = SinkContainer {
+ id: 1,
+ sink: Some(RecordingSink {
+ consumed: Arc::clone(&consumed),
+ }),
+ shutdown: None,
+ };
+
+ let topic_meta = postcard::to_allocvec(&TopicMetadata {
+ stream: "test_stream".to_owned(),
+ topic: "test_topic".to_owned(),
+ })
+ .expect("failed to serialize topic metadata");
+ let messages_meta = postcard::to_allocvec(&MessagesMetadata {
+ partition_id: 0,
+ current_offset: 10_000,
+ schema,
+ })
+ .expect("failed to serialize messages metadata");
+ let messages = postcard::to_allocvec(&RawMessages {
+ schema,
+ messages: vec![RawMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: Vec::new(),
+ payload,
+ }],
+ })
+ .expect("failed to serialize messages");
+
+ // `consume` blocks on the SDK runtime, so this stays a plain test.
+ let status = unsafe {
+ container.consume(
+ topic_meta.as_ptr(),
+ topic_meta.len(),
+ messages_meta.as_ptr(),
+ messages_meta.len(),
+ messages.as_ptr(),
+ messages.len(),
+ )
+ };
+ assert_eq!(status, 0, "the container must accept the run");
+
+ consumed.lock().expect("recorder was poisoned").clone()
+ }
+
+ #[test]
+ fn given_a_proto_tagged_run_when_the_container_consumes_should_hand_the_sink_a_proto_payload() {
+ let recorded = consume_one(Schema::Proto, br#"{"first":"encoded"}"#.to_vec());
+
+ assert_eq!(recorded.len(), 1);
+ let Payload::Proto(text) = &recorded[0] else {
+ panic!("expected a proto payload, got {}", recorded[0]);
+ };
+ assert_eq!(text, r#"{"first":"encoded"}"#);
+ }
+
+ #[test]
+ fn given_a_json_tagged_run_when_the_container_consumes_should_hand_the_sink_a_json_payload() {
+ let recorded = consume_one(Schema::Json, br#"{"first":"encoded"}"#.to_vec());
+
+ assert_eq!(recorded.len(), 1);
+ assert!(
+ matches!(recorded[0], Payload::Json(_)),
+ "got {}",
+ recorded[0]
+ );
+ }
+
+ #[test]
+ fn given_a_text_tagged_run_when_the_container_consumes_should_hand_the_sink_a_text_payload() {
+ let recorded = consume_one(Schema::Text, b"plain text".to_vec());
+
+ assert_eq!(recorded.len(), 1);
+ let Payload::Text(text) = &recorded[0] else {
+ panic!("expected a text payload, got {}", recorded[0]);
+ };
+ assert_eq!(text, "plain text");
+ }
+
+ #[test]
+ fn given_a_raw_tagged_run_when_the_container_consumes_should_hand_the_sink_a_raw_payload() {
+ let recorded = consume_one(Schema::Raw, vec![0xff, 0x00, 0x01]);
+
+ assert_eq!(recorded.len(), 1);
+ let Payload::Raw(bytes) = &recorded[0] else {
+ panic!("expected a raw payload, got {}", recorded[0]);
+ };
+ assert_eq!(bytes, &[0xff, 0x00, 0x01]);
+ }
+}
diff --git a/core/connectors/sdk/src/transforms/flatbuffer_convert.rs b/core/connectors/sdk/src/transforms/flatbuffer_convert.rs
index 5ddc4b170..24d6ec1aa 100644
--- a/core/connectors/sdk/src/transforms/flatbuffer_convert.rs
+++ b/core/connectors/sdk/src/transforms/flatbuffer_convert.rs
@@ -99,18 +99,6 @@ impl FlatBufferConvert {
"unsupported FlatBuffer conversion: {source} -> {target}"
)))
}
-
- fn payload_matches_source_format(source_format: Schema, payload: &Payload) -> bool {
- matches!(
- (source_format, payload),
- (Schema::Json, Payload::Json(_))
- | (Schema::Raw, Payload::Raw(_))
- | (Schema::Text, Payload::Text(_))
- | (Schema::Proto, Payload::Proto(_))
- | (Schema::FlatBuffer, Payload::FlatBuffer(_))
- | (Schema::Avro, Payload::Avro(_))
- )
- }
}
impl Transform for FlatBufferConvert {
@@ -123,7 +111,7 @@ impl Transform for FlatBufferConvert {
_metadata: &TopicMetadata,
mut message: DecodedMessage,
) -> Result<Option<DecodedMessage>, Error> {
- if !Self::payload_matches_source_format(self.config.source_format, &message.payload) {
+ if message.payload.schema() != self.config.source_format {
return Err(Error::InvalidConfigValue(format!(
"expected {} payload",
self.config.source_format
@@ -169,7 +157,14 @@ impl Transform for FlatBufferConvert {
encoder.convert_format(message.payload, Schema::Raw)?
}
(source, target) if source == target => message.payload,
- _ => unreachable!("conversion pair was validated during construction"),
+ // `validate_conversion` rejects every other pair at construction.
+ // Transforms run inside the runtime's consume task, so a panic
+ // here would take the whole connector down rather than one message.
+ (source, target) => {
+ return Err(Error::InvalidConfigValue(format!(
+ "unsupported FlatBuffer conversion: {source} -> {target}"
+ )));
+ }
};
Ok(Some(message))
diff --git a/core/connectors/sdk/src/transforms/proto_convert.rs b/core/connectors/sdk/src/transforms/proto_convert.rs
index 510efa8c0..4f239641f 100644
--- a/core/connectors/sdk/src/transforms/proto_convert.rs
+++ b/core/connectors/sdk/src/transforms/proto_convert.rs
@@ -862,6 +862,57 @@ mod tests {
}
}
+ #[test]
+ fn given_one_configured_instance_when_objects_and_scalars_are_converted_should_emit_different_payload_variants()
+ {
+ // `encode_json_with_schema` only encodes a top-level object, so the
+ // variant this transform returns turns on the message rather than on
+ // the config. A batch carrying both shapes reaches the runtime as a mix
+ // and has to be split into one FFI call per variant.
+ let descriptor = protox_parse::parse(
+ "record.proto",
+ r#"syntax = "proto3";
+ message StringRecord {
+ string first = 1;
+ string second = 2;
+ }"#,
+ )
+ .expect("test schema must parse");
+ let descriptor_set = prost_types::FileDescriptorSet {
+ file: vec![descriptor],
+ };
+ let converter = ProtoConvert::new(ProtoConvertConfig {
+ source_format: Schema::Json,
+ target_format: Schema::Proto,
+ message_type: Some("StringRecord".to_string()),
+ descriptor_set: Some(descriptor_set.encode_to_vec()),
+ ..ProtoConvertConfig::default()
+ });
+
+ for (payload, expected) in [
+ (
+ simd_json::json!({"first": "encoded", "second": "encoded"}),
+ Schema::Raw,
+ ),
+ (simd_json::json!([1, 2, 3]), Schema::Proto),
+ ] {
+ let converted = converter
+ .transform(
+ &create_test_metadata(),
+ create_test_message(Payload::Json(payload)),
+ )
+ .expect("the conversion must not fail")
+ .expect("the message must survive");
+
+ assert_eq!(
+ converted.payload.schema(),
+ expected,
+ "one instance produced {} for a payload expected to be {expected}",
+ converted.payload.schema()
+ );
+ }
+ }
+
fn create_test_message(payload: Payload) -> DecodedMessage {
DecodedMessage {
id: Some(123),
@@ -1093,8 +1144,18 @@ mod tests {
if let Ok(Some(transformed_message)) = result {
if let Payload::Json(json_value) = transformed_message.payload {
if let simd_json::OwnedValue::Object(map) = json_value {
- assert!(map.contains_key("type_url"));
- assert!(map.contains_key("value"));
+ assert_eq!(
+ map.get("type_url").and_then(|value| value.as_str()),
+ Some("type.googleapis.com/google.protobuf.StringValue")
+ );
+ assert_eq!(
+ map.get("value").and_then(|value| value.as_str()),
+ Some(
+ base64::engine::general_purpose::STANDARD
+ .encode(b"Hello, World!")
+ .as_str()
+ )
+ );
} else {
panic!("Expected JSON object");
}
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 1046edb4c..83f30a5ac 100644
--- a/core/connectors/sinks/README.md
+++ b/core/connectors/sinks/README.md
@@ -249,6 +249,14 @@ for message in messages {
While the schema of messages (that will be consumed from the Iggy stream), cannot be controlled by the sink connector itself, the built-in configuration allows to decide what's the expected format of the messages (the particular `StreamDecoder` will be used).
+`messages_metadata.schema` names the variant each `Payload` in the batch actually holds, which is not always the stream's `schema` setting. A decoder may return a different form than the wire format it reads: with `schema = "avro"` the Avro decoder extracts to JSON by default, so the batch arrives as `Payload::Json`. A configured transform can change the variant again. Match on the `Payload` itself and treat `messages_metadata.schema` as a description of what arrived, not of how the stre [...]
+
+One poll can reach the plugin as more than one `consume()` call. The runtime groups a batch into contiguous runs of one payload variant and sends each run on its own, so a batch whose variant changes partway through arrives as several calls.
+
+Every call in that poll repeats the same `messages_metadata.current_offset`, which is the partition's high-water offset from the poll and not the offset of the last message in the call. A sink that keys a commit, a file name or a table version on `current_offset` has to tolerate the repeat.
+
+A batch whose messages were all dropped by the decoder or by a transform still reaches the plugin as one empty call. Nothing is left to read a variant from, so that call carries the stream's configured `schema` instead: an `avro` stream tags its empty batches `avro` and its non-empty ones `json`.
+
Keep in mind, that it might be sometimes difficult/impossible e.g. to transform one format to another e.g. JSON to SBE or so, and in such a case, the consumed messages will be ignored.
Eventually, compile the source code and create a separate connector configuration file in the connectors directory (as specified in the main runtime `config.toml`). Make sure that `path` points to the existing plugin.
diff --git a/core/connectors/sinks/clickhouse_sink/README.md b/core/connectors/sinks/clickhouse_sink/README.md
index 941d3a637..95dc378f9 100644
--- a/core/connectors/sinks/clickhouse_sink/README.md
+++ b/core/connectors/sinks/clickhouse_sink/README.md
@@ -60,7 +60,7 @@ verbose_logging = false
### `json_each_row` (Default)
-Accepts messages with a `Payload::Json` payload. Each payload is serialized on its own line using ClickHouse's `JSONEachRow` format. Send JSON objects whose fields and values are compatible with the existing table and its ClickHouse input settings. The connector does not validate JSON rows against the table schema before sending them.
+Accepts messages with a `Payload::Json` payload, or a `Payload::Proto` payload whose text is a JSON document (what a `proto_convert` transform hands over when it has no descriptor or cannot encode a message). Each document is serialized on its own line using ClickHouse's `JSONEachRow` format. Send JSON objects whose fields and values are compatible with the existing table and its ClickHouse input settings. The connector does not validate JSON rows against the table schema before sending them.
```toml
[plugin_config]
@@ -71,7 +71,7 @@ insert_format = "json_each_row"
### `row_binary`
-Accepts messages with a `Payload::Json` payload. At startup the connector fetches the table schema from `system.columns` and validates that all column types are supported. Messages are then serialised to ClickHouse's `RowBinaryWithDefaults` binary format, which is more efficient than JSON for large volumes.
+Accepts messages with a `Payload::Json` payload, or a `Payload::Proto` payload whose text is a JSON document. At startup the connector fetches the table schema from `system.columns` and validates that all column types are supported. Messages are then serialised to ClickHouse's `RowBinaryWithDefaults` binary format, which is more efficient than JSON for large volumes.
Requires [ClickHouse 23.7 or newer](https://presentations.clickhouse.com/2023-release-23.7/index.html), when `RowBinaryWithDefaults` was introduced. Older servers reject the format; use `json_each_row` instead.
@@ -92,7 +92,7 @@ insert_format = "row_binary"
### `string`
-Accepts messages with a `Payload::Text` payload and appends a newline to each payload that does not already end with one. Set the stream `schema = "text"` and use `string_format` to tell ClickHouse which format to expect.
+Accepts messages with a `Payload::Text` or `Payload::Proto` payload and appends a newline to each payload that does not already end with one. Set the stream `schema = "text"` and use `string_format` to tell ClickHouse which format to expect. Nothing is re-serialized in this mode, so a `proto_convert` transform with `pretty_json = true` combined with `string_format = "json_each_row"` is a misconfiguration: the multi-line document is written as several lines and ClickHouse rejects the insert.
```toml
[plugin_config]
@@ -170,7 +170,7 @@ On shutdown the connector logs the total number of messages processed.
### Bad rows in a batch
-A message whose payload type does not match the chosen format (for example a text payload in JSON mode) is skipped with an error log. The rest of the batch is still sent.
+A message whose payload type does not match the chosen format (for example a text payload in JSON mode, or proto text that is not a JSON document) is skipped with an error log. The rest of the batch is still sent.
The `row_binary` format fails the whole batch on the first JSON row whose values cannot be converted to the target columns. This occurs before any insert request, so the batch does not enter the plugin's insert retry loop. No partial binary row is sent.
diff --git a/core/connectors/sinks/clickhouse_sink/src/body.rs b/core/connectors/sinks/clickhouse_sink/src/body.rs
index c45bdfc91..aa1f04752 100644
--- a/core/connectors/sinks/clickhouse_sink/src/body.rs
+++ b/core/connectors/sinks/clickhouse_sink/src/body.rs
@@ -37,31 +37,34 @@ use tracing::{error, warn};
// ─── Body builders ───────────────────────────────────────────────────────────
/// Build a newline-delimited JSON body for `FORMAT JSONEachRow`.
-/// Each `Payload::Json` message becomes one line. Other payload types are skipped.
+/// Each message holding a JSON document (`Payload::Json`, or `Payload::Proto`
+/// text that parses as JSON) becomes one line. Other payloads are skipped.
+///
+/// Proto text is re-serialised rather than copied through: a `proto_convert`
+/// transform with `pretty_json` set hands over a multi-line document, and
+/// JSONEachRow needs one document per line.
pub(crate) fn build_json_body(messages: &[ConsumedMessage]) -> Vec<u8> {
let mut buf = Vec::with_capacity(messages.len() * 64);
for msg in messages {
- match &msg.payload {
- Payload::Json(value) => {
- if simd_json::to_writer(&mut buf, value).is_ok() {
- buf.push(b'\n');
- } else {
- warn!("Failed to serialise JSON payload at offset {}", msg.offset);
- }
- }
- _ => {
- error!(
- "JSONEachRow mode: skipping unsupported payload type at offset {}",
- msg.offset
- );
- }
+ let Some(value) = msg.payload.json_document() else {
+ error!(
+ "JSONEachRow mode: skipping payload with no JSON document at offset {}",
+ msg.offset
+ );
+ continue;
+ };
+ if simd_json::to_writer(&mut buf, value.as_ref()).is_ok() {
+ buf.push(b'\n');
+ } else {
+ warn!("Failed to serialise JSON payload at offset {}", msg.offset);
}
}
buf
}
/// Build a RowBinaryWithDefaults body.
-/// Each `Payload::Json` message is serialised to binary using the table schema.
+/// Each message holding a JSON document (`Payload::Json`, or `Payload::Proto`
+/// text that parses as JSON) is serialised to binary using the table schema.
///
/// Returns `Err` on the first row that fails to serialise. Unlike the JSON and
/// string builders this cannot skip the bad row: rows are written directly into
@@ -73,24 +76,24 @@ pub(crate) fn build_row_binary_body(
) -> Result<Vec<u8>, Error> {
let mut buf = Vec::with_capacity(messages.len() * 128);
for msg in messages {
- match &msg.payload {
- Payload::Json(value) => {
- crate::binary::serialize_row(value, schema, &mut buf)?;
- }
- _ => {
- error!(
- "RowBinary mode: skipping unsupported payload type at offset {}",
- msg.offset
- );
- }
- }
+ let Some(value) = msg.payload.json_document() else {
+ error!(
+ "RowBinary mode: skipping payload with no JSON document at offset {}",
+ msg.offset
+ );
+ continue;
+ };
+ crate::binary::serialize_row(value.as_ref(), schema, &mut buf)?;
}
Ok(buf)
}
/// Build a raw string body for CSV / TSV / JSONEachRow string passthrough.
-/// Each `Payload::Text` message is written as-is with a trailing newline
-/// appended if not already present.
+/// Each `Payload::Text` or `Payload::Proto` message is written as-is with a
+/// trailing newline appended if not already present. Nothing is re-serialised
+/// here, so a `proto_convert` transform with `pretty_json` set combined with
+/// the `json_each_row` string format is a misconfiguration: the multi-line
+/// document is written as several lines.
pub(crate) fn build_string_body(
messages: &[ConsumedMessage],
string_format: StringFormat,
@@ -98,7 +101,7 @@ pub(crate) fn build_string_body(
let mut buf = Vec::with_capacity(messages.len() * 64);
for msg in messages {
match &msg.payload {
- Payload::Text(s) => {
+ Payload::Text(s) | Payload::Proto(s) => {
buf.extend_from_slice(s.as_bytes());
if string_format.requires_newline() && !s.ends_with('\n') {
buf.push(b'\n');
@@ -280,6 +283,38 @@ mod tests {
assert!(build_string_body(&messages, StringFormat::Csv).is_empty());
}
+ /// A `proto_convert` transform that cannot encode falls back to proto text,
+ /// and the runtime tags that run `Schema::Proto`. Passthrough mode takes it
+ /// as text, the way it reached this sink before batches were tagged from
+ /// the payload.
+ #[test]
+ fn string_body_proto_payload_is_written_as_text() {
+ let messages = vec![msg(Payload::Proto("a,b,c".to_owned()))];
+ assert_eq!(build_string_body(&messages, StringFormat::Csv), b"a,b,c\n");
+ }
+
+ /// A `proto_convert` transform with no descriptor hands over the JSON it
+ /// was given as proto text, so the row is the document that text holds.
+ #[test]
+ fn json_body_proto_payload_holding_json_is_written() {
+ let messages = vec![msg(Payload::Proto(r#"{"name":"alice"}"#.into()))];
+ assert_eq!(build_json_body(&messages), b"{\"name\":\"alice\"}\n");
+ }
+
+ /// `pretty_json` puts newlines in the proto text and JSONEachRow needs one
+ /// document per line, so the document is re-serialised rather than copied.
+ #[test]
+ fn json_body_pretty_proto_json_is_written_as_one_line() {
+ let messages = vec![msg(Payload::Proto("{\n \"name\": \"alice\"\n}".into()))];
+ assert_eq!(build_json_body(&messages), b"{\"name\":\"alice\"}\n");
+ }
+
+ #[test]
+ fn json_body_non_json_proto_payload_is_skipped() {
+ let messages = vec![msg(Payload::Proto("name: \"alice\"".into()))];
+ assert!(build_json_body(&messages).is_empty());
+ }
+
// ── build_row_binary_body ────────────────────────────────────────────────
#[test]
@@ -310,6 +345,21 @@ mod tests {
assert_eq!(body, b"\x00\x05alice");
}
+ #[test]
+ fn row_binary_body_proto_payload_holding_json_writes_bytes() {
+ let messages = vec![msg(Payload::Proto(r#"{"name":"alice"}"#.into()))];
+ let schema = vec![col("name", ChType::String)];
+ let body = build_row_binary_body(&messages, &schema).unwrap();
+ assert_eq!(body, b"\x00\x05alice");
+ }
+
+ #[test]
+ fn row_binary_body_non_json_proto_payload_is_skipped() {
+ let messages = vec![msg(Payload::Proto("name: \"alice\"".into()))];
+ let body = build_row_binary_body(&messages, &[col("name", ChType::String)]).unwrap();
+ assert!(body.is_empty());
+ }
+
#[test]
fn row_binary_body_multiple_rows_concatenated() {
let mut obj1 = simd_json::owned::Object::with_capacity(1);
diff --git a/core/connectors/sinks/delta_sink/Cargo.toml b/core/connectors/sinks/delta_sink/Cargo.toml
index ea544287c..0c36390b2 100644
--- a/core/connectors/sinks/delta_sink/Cargo.toml
+++ b/core/connectors/sinks/delta_sink/Cargo.toml
@@ -48,5 +48,8 @@ tokio = { workspace = true }
tracing = { workspace = true }
url = { workspace = true }
+[dev-dependencies]
+simd-json = { workspace = true }
+
[lints]
workspace = true
diff --git a/core/connectors/sinks/delta_sink/src/sink.rs b/core/connectors/sinks/delta_sink/src/sink.rs
index e93b97f04..4bba68853 100644
--- a/core/connectors/sinks/delta_sink/src/sink.rs
+++ b/core/connectors/sinks/delta_sink/src/sink.rs
@@ -22,8 +22,7 @@ use crate::storage::build_storage_options;
use async_trait::async_trait;
use deltalake::writer::{DeltaWriter, JsonWriter};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
- owned_value_to_serde_json,
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, owned_value_to_serde_json,
};
use tracing::{debug, error, info};
@@ -101,21 +100,7 @@ impl Sink for DeltaSink {
messages_metadata.current_offset,
);
- // Extract JSON values from consumed messages
- let mut json_values: Vec<serde_json::Value> = Vec::with_capacity(messages.len());
- for msg in &messages {
- match &msg.payload {
- Payload::Json(simd_value) => {
- json_values.push(owned_value_to_serde_json(simd_value));
- }
- other => {
- error!(
- "Unsupported payload type: {other}. Delta sink only supports JSON payloads."
- );
- return Err(Error::InvalidPayloadType);
- }
- }
- }
+ let mut json_values = json_values(&messages)?;
if json_values.is_empty() {
debug!("No JSON values to write");
@@ -178,3 +163,65 @@ impl Sink for DeltaSink {
Ok(())
}
}
+
+/// Every message's JSON document as a serde value, or the batch error for a
+/// payload that has none. Proto text holding JSON is the descriptor-less
+/// `proto_convert` fallback and is written as the document it holds; proto
+/// text that is not JSON fails the batch like any other non-JSON payload.
+fn json_values(messages: &[ConsumedMessage]) -> Result<Vec<serde_json::Value>, Error> {
+ let mut json_values = Vec::with_capacity(messages.len());
+ for message in messages {
+ let Some(document) = message.payload.json_document() else {
+ error!(
+ "Unsupported payload type: {}. Delta sink only supports JSON payloads.",
+ message.payload
+ );
+ return Err(Error::InvalidPayloadType);
+ };
+ json_values.push(owned_value_to_serde_json(document.as_ref()));
+ }
+ Ok(json_values)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::json_values;
+ use iggy_connector_sdk::{ConsumedMessage, Error, Payload};
+
+ fn message(payload: Payload) -> ConsumedMessage {
+ ConsumedMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: None,
+ payload,
+ }
+ }
+
+ #[test]
+ fn json_values_parses_proto_text_that_is_json() {
+ let messages = vec![
+ message(Payload::Json(simd_json::json!({"id": 1}))),
+ message(Payload::Proto(r#"{"id": 2}"#.to_owned())),
+ ];
+
+ let values = json_values(&messages).expect("proto text holding JSON is a document");
+
+ assert_eq!(
+ values,
+ vec![serde_json::json!({"id": 1}), serde_json::json!({"id": 2})]
+ );
+ }
+
+ #[test]
+ fn json_values_rejects_proto_text_that_is_not_json() {
+ let messages = vec![message(Payload::Proto("id: 2".to_owned()))];
+
+ assert_eq!(
+ json_values(&messages).expect_err("proto text that is not JSON has no document"),
+ Error::InvalidPayloadType
+ );
+ }
+}
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index cc4bec450..2c786e1d2 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -7,6 +7,7 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
- The target Doris **database and table must be pre-created** before enabling the sink. The connector never issues DDL.
- `database` and `table` config values must match `[A-Za-z0-9_]+`. Anything else is rejected at startup with `Error::InvalidConfigValue` — this also prevents path traversal in the constructed `/api/{db}/{table}/_stream_load` URL.
- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). If a non-JSON payload reaches the connector it logs at `error!` and aborts the whole poll; since the consumer offset is already committed at poll time the batch is not replayed - data loss despite the error log - so the upstream schema must be guaranteed JSON. (Under `schema = "json"` the SDK drops non-JSON before the connector sees it, so this abort is a defensive guard.)
+- That guard is reachable in one case: a format-converting transform runs after the decoder and can hand the connector a non-JSON variant even under `schema = "json"`. `Payload::Proto` is the exception: a `proto_convert` transform with no descriptor hands over the JSON it was given as proto text, and the connector loads that as the document it holds. Proto text that is not JSON still aborts the poll.
- The Iggy message JSON shape must match the target table columns. JSON fields map by name. Use the optional `columns` plugin setting for field mappings or derived expressions.
## How it works
@@ -105,7 +106,7 @@ timeout = "30s"
- **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it.
- **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load (with its own in-request retry budget for transient failures). If a chunk still fails after its retries (serialize, HTTP, or status-classification error), the connector keeps the first error, attempts the remaining chunks, and returns that error at the end — it does **not** stop at the first such failure.
The runtime commits the consumer offset for the whole poll before `consume()` runs, so a chunk that exhausts its in-request retries is not replayed across polls; pushing the other chunks through maximizes delivered data, and the first error is surfaced at the end (logged at `error!` for observability; the runtime counts the error and continues polling, with no automatic replay of the failed poll or DLQ).
- The one deliberate exception is a **non-JSON payload**, which is treated as a schema-contract violation and aborts the whole poll immediately (see the Requirements note above). Under `schema = "json"` this is unreachable, so it is a defensive guard rather than a normal path.
+ The one deliberate exception is a **non-JSON payload**, which is treated as a schema-contract violation and aborts the whole poll immediately (see the Requirements note above). Under `schema = "json"` with no transforms configured this is unreachable, so it is a defensive guard rather than a normal path. Proto text holding a JSON document does not trigger it.
## Limitations
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index 9ed094876..de8a83800 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -21,13 +21,14 @@ use bytes::Bytes;
use humantime::Duration as HumanDuration;
use iggy_connector_sdk::retry::{RetryPolicy, retry_async};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Schema, Sink, TopicMetadata, sink_connector,
};
use reqwest::{Method, StatusCode, header};
use secrecy::zeroize::Zeroizing;
use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
use simd_json::{OwnedValue, StaticNode};
+use std::borrow::Cow;
use std::io::Write as _;
use std::str::FromStr;
use std::time::Duration;
@@ -639,6 +640,28 @@ fn csv_column_names(columns: &str) -> Vec<String> {
.collect()
}
+/// The JSON document of every message in a chunk, or the schema-contract error
+/// that aborts the poll. Proto text is read as the document it holds, so the
+/// descriptor-less `proto_convert` fallback lands, and proto text that is not
+/// JSON is rejected like any other payload without a document.
+fn json_documents(
+ chunk: &[ConsumedMessage],
+ sink_id: u32,
+ schema: Schema,
+) -> Result<Vec<Cow<'_, OwnedValue>>, Error> {
+ chunk
+ .iter()
+ .map(|message| {
+ message.payload.json_document().ok_or_else(|| {
+ error!(
+ "Doris sink ID {sink_id} received payload with no JSON document (schema={schema}); aborting poll"
+ );
+ Error::InvalidPayloadType
+ })
+ })
+ .collect()
+}
+
/// Serialize a chunk into the request body for the configured Stream Load format.
fn build_request_body(
format: Format,
@@ -1096,24 +1119,19 @@ impl Sink for DorisSink {
// boundary (severity isn't propagated), and every chunk error is already
// logged individually below — so first-error is sufficient.
//
- // The lone hard-abort is a non-JSON payload (via `?`): a stream-wide
- // schema-contract violation, not a transient chunk failure. Under the
- // documented `schema = "json"` config the SDK drops non-JSON before
- // consume() is called, so this stands as a defensive guard.
+ // The lone hard-abort is a payload with no JSON document (via `?`): a
+ // stream-wide schema-contract violation, not a transient chunk failure.
+ // Under the documented `schema = "json"` config the runtime's JSON
+ // decoder drops non-JSON bytes before consume() is called, so this
+ // stands as a defensive guard unless a format-converting transform is
+ // configured. `proto_convert` is the one that reaches it: with no
+ // descriptor it hands over the JSON it was given as proto text, which
+ // is taken as the document it holds, and proto text that is not JSON
+ // still aborts.
for chunk in messages.chunks(batch_size) {
- let json_values: Vec<&simd_json::OwnedValue> = chunk
- .iter()
- .map(|m| match &m.payload {
- Payload::Json(value) => Ok(value),
- _ => {
- error!(
- "Doris sink ID {} received non-JSON payload (schema={}); aborting poll",
- self.id, messages_metadata.schema
- );
- Err(Error::InvalidPayloadType)
- }
- })
- .collect::<Result<_, _>>()?;
+ let documents = json_documents(chunk, self.id, messages_metadata.schema)?;
+ let json_values: Vec<&simd_json::OwnedValue> =
+ documents.iter().map(Cow::as_ref).collect();
// `chunks()` never yields an empty slice, so first/last are present.
// Use `zip` + `continue` (not `.expect`) so that if a future refactor
@@ -1202,6 +1220,7 @@ impl Sink for DorisSink {
#[cfg(test)]
mod tests {
use super::*;
+ use iggy_connector_sdk::Payload;
fn make_config() -> DorisSinkConfig {
DorisSinkConfig {
@@ -2346,6 +2365,38 @@ mod tests {
assert!(csv_column_names("calc = x").is_empty());
}
+ fn proto_message(text: &str) -> ConsumedMessage {
+ ConsumedMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Proto(text.to_string()),
+ }
+ }
+
+ #[test]
+ fn json_documents_reads_proto_text_that_is_json() {
+ let chunk = vec![proto_message(r#"{"id":1}"#)];
+
+ let documents = json_documents(&chunk, 1, Schema::Proto).unwrap();
+
+ assert_eq!(documents.len(), 1);
+ assert_eq!(*documents[0], owned(r#"{"id":1}"#));
+ }
+
+ #[test]
+ fn json_documents_rejects_proto_text_that_is_not_json() {
+ let chunk = vec![proto_message("id: 1")];
+
+ assert_eq!(
+ json_documents(&chunk, 1, Schema::Proto).unwrap_err(),
+ Error::InvalidPayloadType
+ );
+ }
+
#[test]
fn build_csv_body_encodes_scalars_nulls_and_missing() {
let row = owned(r#"{"i":-5,"u":10,"f":2.5,"b":false,"nul":null,"empty":""}"#);
diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
index 438131fca..f4ff03e07 100644
--- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs
+++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
@@ -25,7 +25,7 @@ use elasticsearch::{
use iggy_common::IggyTimestamp;
use iggy_connector_sdk::retry::{RetryPolicy, parse_duration};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink, TopicMetadata, sink_connector,
};
use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
@@ -312,6 +312,32 @@ impl ElasticsearchSink {
}
}
+/// The document a payload indexes as. `None` skips the message.
+fn document_from_payload(payload: Payload, schema: Schema) -> Option<OwnedValue> {
+ let payload = payload.into_json_document();
+ match payload {
+ Payload::Json(value) => Some(value),
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy) {
+ Ok(value) => Some(value),
+ Err(_) => Some(simd_json::json!({
+ "data": general_purpose::STANDARD.encode(&bytes),
+ "data_type": "raw"
+ })),
+ }
+ }
+ Payload::Text(text) | Payload::Proto(text) => Some(simd_json::json!({
+ "text": text,
+ "data_type": "text"
+ })),
+ _ => {
+ warn!("Unsupported payload format: {schema}");
+ None
+ }
+ }
+}
+
#[async_trait]
impl Sink for ElasticsearchSink {
async fn open(&mut self) -> Result<(), Error> {
@@ -367,28 +393,9 @@ impl Sink for ElasticsearchSink {
let messages_count = messages.len();
let mut documents = Vec::with_capacity(messages_count);
for message in messages {
- let mut doc = match message.payload {
- Payload::Json(value) => value,
- Payload::Raw(bytes) => {
- let mut bytes_copy = bytes.clone();
- match simd_json::from_slice::<OwnedValue>(&mut bytes_copy) {
- Ok(value) => value,
- Err(_) => {
- simd_json::json!({
- "data": general_purpose::STANDARD.encode(&bytes),
- "data_type": "raw"
- })
- }
- }
- }
- Payload::Text(text) => simd_json::json!({
- "text": text,
- "data_type": "text"
- }),
- _ => {
- warn!("Unsupported payload format: {}", messages_metadata.schema);
- continue;
- }
+ let Some(mut doc) = document_from_payload(message.payload, messages_metadata.schema)
+ else {
+ continue;
};
// Add metadata fields
@@ -492,6 +499,33 @@ mod tests {
assert_eq!(sink.retry_policy.max_delay, Duration::from_secs(5));
}
+ #[test]
+ fn given_proto_text_holding_json_when_building_a_document_should_keep_the_original_fields() {
+ let payload = Payload::Proto(r#"{"name":"user_1","amount":2.5}"#.to_owned());
+
+ let document = document_from_payload(payload, Schema::Proto)
+ .expect("proto text holding JSON is a document");
+
+ assert_eq!(
+ document,
+ simd_json::json!({"name": "user_1", "amount": 2.5}),
+ "the document must keep its fields rather than become a text blob"
+ );
+ }
+
+ #[test]
+ fn given_proto_text_that_is_not_json_when_building_a_document_should_index_it_as_text() {
+ let payload = Payload::Proto("name: \"user_1\"".to_owned());
+
+ let document =
+ document_from_payload(payload, Schema::Proto).expect("proto text still indexes");
+
+ assert_eq!(
+ document,
+ simd_json::json!({"text": "name: \"user_1\"", "data_type": "text"})
+ );
+ }
+
#[test]
fn given_configured_retry_delays_when_loading_should_parse_or_fall_back() {
for (delay, max_delay, expected_delay, expected_max) in [
diff --git a/core/connectors/sinks/http_sink/README.md b/core/connectors/sinks/http_sink/README.md
index f8d9dd8b8..97a969509 100644
--- a/core/connectors/sinks/http_sink/README.md
+++ b/core/connectors/sinks/http_sink/README.md
@@ -236,7 +236,8 @@ When `include_metadata = true` (default), payloads are wrapped:
```
- **`iggy_id`**: Message ID formatted as 32-character lowercase hex string (no dashes)
-- **Non-JSON payload variants** (Raw, FlatBuffer, Proto, Avro): base64-encoded with `"iggy_payload_encoding": "base64"` in payload
+- **Proto**: sent as the JSON document it holds when the text parses as JSON (the descriptor-less `proto_convert` fallback), otherwise as a plain string like Text
+- **Binary payload variants** (Raw, FlatBuffer, Avro): base64-encoded with `"iggy_payload_encoding": "base64"` in payload
- **JSON/Text payloads**: JSON values are reserialized; text becomes a JSON string.
- **Timestamps**: `iggy_timestamp` and optional `iggy_origin_timestamp` use Unix epoch microseconds.
- **Optional metadata**: `include_checksum` and `include_origin_timestamp` default to false. Nonempty message headers are included as `iggy_headers`.
diff --git a/core/connectors/sinks/http_sink/src/lib.rs b/core/connectors/sinks/http_sink/src/lib.rs
index 1b4f8d306..7a1c226db 100644
--- a/core/connectors/sinks/http_sink/src/lib.rs
+++ b/core/connectors/sinks/http_sink/src/lib.rs
@@ -372,18 +372,20 @@ impl HttpSink {
}
/// Convert a `Payload` to a JSON value for metadata wrapping.
- /// Non-JSON payloads are base64-encoded with a `iggy_payload_encoding` marker.
+ /// Text payloads, and proto text that is not JSON, are sent as strings; binary
+ /// payloads are base64-encoded with a `iggy_payload_encoding` marker.
///
/// Note: All current `Payload` variants produce infallible conversions.
/// The `Result` return type exists as a safety net for future variants.
fn payload_to_json(&self, payload: Payload) -> Result<serde_json::Value, Error> {
+ let payload = payload.into_json_document();
match payload {
Payload::Json(value) => {
// Direct structural conversion (not serialization roundtrip).
// Follows the Elasticsearch sink pattern. NaN/Infinity f64 → null.
Ok(owned_value_to_serde_json(&value))
}
- Payload::Text(text) => Ok(serde_json::Value::String(text)),
+ Payload::Text(text) | Payload::Proto(text) => Ok(serde_json::Value::String(text)),
Payload::Raw(bytes) | Payload::FlatBuffer(bytes) => {
let encoded = EncodedPayload {
data: general_purpose::STANDARD.encode(&bytes),
@@ -392,14 +394,6 @@ impl HttpSink {
serde_json::to_value(encoded)
.map_err(|e| Error::Serialization(format!("EncodedPayload: {}", e)))
}
- Payload::Proto(proto_str) => {
- let encoded = EncodedPayload {
- data: general_purpose::STANDARD.encode(proto_str.as_bytes()),
- iggy_payload_encoding: ENCODING_BASE64,
- };
- serde_json::to_value(encoded)
- .map_err(|e| Error::Serialization(format!("EncodedPayload: {}", e)))
- }
Payload::Avro(bytes) => {
let encoded = EncodedPayload {
data: general_purpose::STANDARD.encode(&bytes),
@@ -1194,8 +1188,10 @@ impl Sink for HttpSink {
///
/// **Runtime note**: The FFI boundary in `sdk/src/sink.rs` maps `consume()`'s `Result` to
/// `i32` (0=ok, 1=err), but the runtime's `process_messages()` in `runtime/src/sink.rs`
- /// logs and counts a nonzero code, records no processed messages for that batch,
- /// and continues polling. Returning `Err` does not trigger a runtime retry.
+ /// logs and counts a nonzero code, records no processed messages for the run that
+ /// returned it, and continues polling. A batch that mixes payload variants is split
+ /// into one `consume()` call per run, so the other runs of that poll still count as
+ /// processed. Returning `Err` does not trigger a runtime retry.
async fn consume(
&self,
topic_metadata: &TopicMetadata,
@@ -1552,16 +1548,21 @@ mod tests {
}
#[test]
- fn given_proto_payload_should_base64_encode_string_bytes() {
+ fn given_proto_payload_holding_json_should_send_the_document() {
+ let sink = given_sink_with_defaults();
+ let result = sink
+ .payload_to_json(Payload::Proto(r#"{"id":1,"name":"row-1"}"#.to_string()))
+ .unwrap();
+ assert_eq!(result, serde_json::json!({"id": 1, "name": "row-1"}));
+ }
+
+ #[test]
+ fn given_proto_text_that_is_not_json_should_send_it_as_a_string() {
let sink = given_sink_with_defaults();
let result = sink
.payload_to_json(Payload::Proto("proto_data".to_string()))
.unwrap();
- assert_eq!(result[FIELD_PAYLOAD_ENCODING], "base64");
- assert_eq!(
- result[FIELD_DATA],
- general_purpose::STANDARD.encode(b"proto_data")
- );
+ assert_eq!(result, serde_json::Value::String("proto_data".to_string()));
}
#[test]
diff --git a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
index 442705c5e..17a717bd9 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
@@ -83,19 +83,25 @@ impl DynamicRouter {
}
fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String> {
- match &message.payload {
- Payload::Json(payload) => payload
- .as_object()
- .and_then(|obj| obj.get(&self.route_field))
- .map(|val| val.to_string()),
- _ => {
- warn!("Unsupported format for iceberg connector");
- None
- }
- }
+ route_value(&message.payload, &self.route_field)
}
}
+/// The routing value a payload carries under `route_field`, read from the JSON
+/// document it holds. `route_data` normalises the payload first, so proto text
+/// holding JSON arrives here as `Payload::Json`; proto text that is not JSON
+/// holds no document and does not route, and neither does any other variant.
+fn route_value(payload: &Payload, route_field: &str) -> Option<String> {
+ let Some(document) = payload.json_document() else {
+ warn!("Unsupported format for iceberg connector");
+ return None;
+ };
+ document
+ .as_object()
+ .and_then(|obj| obj.get(route_field))
+ .map(|val| val.to_string())
+}
+
#[async_trait]
impl Router for DynamicRouter {
async fn route_data(
@@ -104,7 +110,10 @@ impl Router for DynamicRouter {
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error> {
let mut writer = DynamicWriter::new();
- for message in messages {
+ for mut message in messages {
+ // Normalised once here so the routing read and the later write both
+ // work off the document, rather than parsing the same proto text twice.
+ message.payload = message.payload.into_json_document();
let route_field_val = match self.extract_route_field(&message) {
Some(val) => val,
None => continue,
@@ -165,3 +174,30 @@ impl Router for DynamicRouter {
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::route_value;
+ use iggy_connector_sdk::Payload;
+
+ #[test]
+ fn given_proto_text_holding_json_should_route_by_the_route_field() {
+ let payload = Payload::Proto(r#"{"id": 1, "region": "eu"}"#.to_owned());
+
+ assert_eq!(route_value(&payload, "region").as_deref(), Some("eu"));
+ }
+
+ #[test]
+ fn given_proto_text_that_is_not_json_should_not_route() {
+ let payload = Payload::Proto("region: \"eu\"".to_owned());
+
+ assert_eq!(route_value(&payload, "region"), None);
+ }
+
+ #[test]
+ fn given_a_json_payload_without_the_route_field_should_not_route() {
+ let payload = Payload::Json(simd_json::json!({ "id": 1 }));
+
+ assert_eq!(route_value(&payload, "region"), None);
+ }
+}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
index 14b19717b..43fecfbfd 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
@@ -36,6 +36,7 @@ use iceberg::{
};
use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
use parquet::file::properties::WriterProperties;
+use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use tracing::{error, warn};
@@ -146,21 +147,24 @@ async fn write_data_files(
let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
- let msgs: Vec<&simd_json::OwnedValue> = messages
+ // Proto text holding JSON is the descriptor-less `proto_convert` fallback
+ // and is written as the document it holds. Proto text that is not JSON is
+ // skipped like any other payload without a document.
+ let documents: Vec<Cow<'_, simd_json::OwnedValue>> = messages
.iter()
- .filter_map(|payload| match payload {
- Payload::Json(value) => Some(value),
- _ => {
+ .filter_map(|payload| {
+ let document = payload.json_document();
+ if document.is_none() {
warn!(
"Unsupported type of payload, expected JSON, got {}",
messages_schema.to_string()
);
- None
}
+ document
})
.collect();
- if msgs.is_empty() {
+ if documents.is_empty() {
error!(
"Batch of {} messages has no JSON payloads, the Iceberg sink requires schema = json",
messages.len()
@@ -168,6 +172,7 @@ async fn write_data_files(
return Err(Error::InvalidPayloadType);
}
+ let msgs: Vec<&simd_json::OwnedValue> = documents.iter().map(Cow::as_ref).collect();
let cursor = JsonArrowReader::new(msgs.as_slice());
let reader = ReaderBuilder::new(Arc::new(
schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| {
@@ -439,6 +444,32 @@ mod tests {
}
}
+ #[test]
+ fn given_proto_text_payloads_holding_json_should_write_them_as_rows() {
+ let table = in_memory_table(UnboundPartitionSpec::builder().build());
+ let payloads = vec![
+ Payload::Json(simd_json::json!({ "id": 1, "region": "eu" })),
+ Payload::Proto(r#"{"id": 2, "region": "us"}"#.to_owned()),
+ ];
+
+ let data_files = write(&table, &payloads);
+
+ assert_eq!(data_files.len(), 1);
+ assert_eq!(data_files[0].record_count(), 2);
+ }
+
+ #[test]
+ fn given_only_proto_text_that_is_not_json_should_fail_with_invalid_payload_type() {
+ let table = in_memory_table(UnboundPartitionSpec::builder().build());
+ let payloads = vec![Payload::Proto("id: 1".to_owned())];
+
+ let error = test_runtime()
+ .block_on(write_data_files(&payloads, &table, Schema::Proto))
+ .expect_err("proto text that is not JSON has no document to write");
+
+ assert_eq!(error, Error::InvalidPayloadType);
+ }
+
#[test]
fn given_partitioned_table_should_write_one_data_file_per_partition_value() {
let partition_spec = UnboundPartitionSpec::builder()
diff --git a/core/connectors/sinks/influxdb_sink/dependencies.md b/core/connectors/sinks/influxdb_sink/dependencies.md
index 63fd1d596..b056d46e4 100644
--- a/core/connectors/sinks/influxdb_sink/dependencies.md
+++ b/core/connectors/sinks/influxdb_sink/dependencies.md
@@ -41,7 +41,7 @@ to `cargo tree -p iggy_connector_influxdb_sink` for the full graph.
| `base64` | `^0.22.1` | MIT / Apache-2.0 | Encodes raw message payloads as base64 when `payload_format = "base64"` is configured. Uses `Engine::encode_string` to write directly into the line-protocol output buffer with no intermediate allocation. |
| `bytes` | `^1.11.1` | MIT | Zero-copy `Bytes::from(body.into_bytes())` converts the line-protocol string body into the request body sent to InfluxDB's write endpoint without an extra copy. |
| `iggy_common` | `^0.11.0` | Apache-2.0 | Shared Iggy types: `serde_secret` for safe token serialisation in config structs. |
-| `iggy_connector_sdk` | `^0.4.0` | Apache-2.0 | Core connector abstractions: `Sink` trait, `ConsumedMessage`, `MessagesMetadata`, `TopicMetadata`, `Error`, retry/circuit-breaker utilities, and the `sink_connector!` registration macro. |
+| `iggy_connector_sdk` | `^0.5.0` | Apache-2.0 | Core connector abstractions: `Sink` trait, `ConsumedMessage`, `MessagesMetadata`, `TopicMetadata`, `Error`, retry/circuit-breaker utilities, and the `sink_connector!` registration macro. |
| `reqwest` | `^0.13.3` | MIT / Apache-2.0 | Async HTTP client used to POST line-protocol batches to `/api/v2/write` (V2) and `/api/v3/write_lp` (V3). |
| `reqwest-middleware` | `^0.5.1` | MIT | Middleware wrapper around `reqwest::Client` that attaches the retry and tracing layers built by `iggy_connector_sdk::retry::build_retry_client`. |
| `secrecy` | `^0.10` | MIT / Apache-2.0 | `SecretString` / `SecretBox` wrappers that prevent accidental logging of API tokens in config structs and the `auth_header` field. |
diff --git a/core/connectors/sinks/meilisearch_sink/README.md b/core/connectors/sinks/meilisearch_sink/README.md
index 8545198c6..aa49ea378 100644
--- a/core/connectors/sinks/meilisearch_sink/README.md
+++ b/core/connectors/sinks/meilisearch_sink/README.md
@@ -59,11 +59,14 @@ max_open_retries = 5
JSON object payloads are indexed as documents. JSON arrays or scalar values are
wrapped in a `value` field because Meilisearch documents must be objects. Raw
payloads are parsed as JSON when possible; otherwise, they are indexed as base64
-data. Text payloads are indexed in a `text` field. Unsupported payload schemas
-are skipped with a warning and counted in the plugin's private error counter.
-The callback returns success after these drops, so runtime statistics can count
-those records as processed. Offsets are auto-committed while polling, before
-indexing completes. There is no built-in dead-letter queue for these drops.
+data. Text payloads are indexed in a `text` field. Proto payloads holding a JSON
+document, which is what the descriptor-less `proto_convert` fallback produces,
+are indexed as that document; proto text that is not JSON is indexed in the
+`text` field alongside text payloads. Avro and FlatBuffer payloads are skipped
+with a warning and counted in the plugin's private error counter. The callback
+returns success after these drops, so runtime statistics can count those records
+as processed. Offsets are auto-committed while polling, before indexing
+completes. There is no built-in dead-letter queue for these drops.
When the configured primary key is absent, the connector injects a stable value
derived from the exact Iggy stream, topic, partition, offset, and message ID.
diff --git a/core/connectors/sinks/meilisearch_sink/src/lib.rs b/core/connectors/sinks/meilisearch_sink/src/lib.rs
index 5f6be75db..3f56cea57 100644
--- a/core/connectors/sinks/meilisearch_sink/src/lib.rs
+++ b/core/connectors/sinks/meilisearch_sink/src/lib.rs
@@ -339,6 +339,7 @@ impl MeilisearchSink {
payload,
} = message;
+ let payload = payload.into_json_document();
let mut document = match payload {
Payload::Json(value) => {
Self::document_from_json_value(owned_value_into_serde_json(value))
@@ -360,7 +361,7 @@ impl MeilisearchSink {
]),
}
}
- Payload::Text(text) => Map::from_iter([
+ Payload::Text(text) | Payload::Proto(text) => Map::from_iter([
("text".to_string(), Value::String(text)),
("data_type".to_string(), Value::String("text".to_string())),
]),
@@ -1182,6 +1183,41 @@ mod tests {
}
}
+ #[test]
+ fn proto_payloads_holding_json_are_indexed_as_documents() {
+ let sink = sink_with_config(base_config());
+ let message = message(Payload::Proto(r#"{"id":1,"name":"row-1"}"#.to_owned()));
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(), message)
+ .expect("proto text holding JSON is a document");
+
+ assert_eq!(
+ document.get("name"),
+ Some(&Value::String("row-1".to_string()))
+ );
+ assert_eq!(document.get("data_type"), None);
+ }
+
+ #[test]
+ fn proto_payloads_that_are_not_json_are_stored_as_text() {
+ let sink = sink_with_config(base_config());
+ let message = message(Payload::Proto("name: \"row-1\"".to_owned()));
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(), message)
+ .expect("proto text still indexes");
+
+ assert_eq!(
+ document.get("text"),
+ Some(&Value::String("name: \"row-1\"".to_string()))
+ );
+ assert_eq!(
+ document.get("data_type"),
+ Some(&Value::String("text".to_string()))
+ );
+ }
+
#[test]
fn swaps_retry_delays_when_retry_delay_exceeds_max_retry_delay() {
let mut config = base_config();
diff --git a/core/connectors/sinks/quickwit_sink/README.md b/core/connectors/sinks/quickwit_sink/README.md
index 40b5971d6..b64575721 100644
--- a/core/connectors/sinks/quickwit_sink/README.md
+++ b/core/connectors/sinks/quickwit_sink/README.md
@@ -97,7 +97,7 @@ The stream's `schema` selects the runtime decoder. The sink sends JSON objects u
| Raw non-UTF-8 bytes | `{"data":"/wCA","data_type":"raw","data_encoding":"base64"}` |
| Text | `{"text":"ready","data_type":"text"}` |
-Raw JSON arrays and scalars remain UTF-8 strings in the raw wrapper. Malformed JSON also preserves the original bytes. `data_encoding` distinguishes literal text from base64. The sink handles `Payload::Avro` and `Payload::FlatBuffer` with the raw path, and `Payload::Proto` with the text wrapper. Runtime decoder settings determine which payload variant reaches the sink.
+Raw JSON arrays and scalars remain UTF-8 strings in the raw wrapper. Malformed JSON also preserves the original bytes. `data_encoding` distinguishes literal text from base64. The sink handles `Payload::Avro` and `Payload::FlatBuffer` with the raw path. `Payload::Proto` whose text is a JSON document (the descriptor-less `proto_convert` fallback) takes the JSON path; any other proto text takes the text wrapper. Runtime decoder settings determine which payload variant reaches the sink.
The examples use `mode: dynamic` to retain wrapper fields. With `mode: strict`, map every field emitted by the selected payload shape or Quickwit rejects the document during indexing, even after a successful HTTP response. Timestamp sharding and retention are optional here: raw/text wrappers have no `timestamp`, and the `add_fields` transform only enriches JSON payloads. Configure them only when every document supplies the required timestamp.
diff --git a/core/connectors/sinks/quickwit_sink/src/lib.rs b/core/connectors/sinks/quickwit_sink/src/lib.rs
index 61e013a90..446da8895 100644
--- a/core/connectors/sinks/quickwit_sink/src/lib.rs
+++ b/core/connectors/sinks/quickwit_sink/src/lib.rs
@@ -313,7 +313,8 @@ impl QuickwitSink {
fn extract_json_payloads(&self, messages: Vec<ConsumedMessage>) -> Vec<OwnedValue> {
let mut json_payloads = Vec::with_capacity(messages.len());
for message in messages {
- let val = match message.payload {
+ let payload = message.payload.into_json_document();
+ let val = match payload {
Payload::Json(value @ OwnedValue::Object(_)) => value,
Payload::Json(value) => simd_json::json!({
"data": value,
@@ -661,6 +662,25 @@ mod tests {
assert_eq!(&extracted[4..], &[text.clone(), text]);
}
+ #[test]
+ fn given_proto_text_holding_json_when_extracted_should_preserve_the_document() {
+ let sink = QuickwitSink::new(1, test_config());
+ let messages = vec![
+ test_message(Payload::Proto(r#"{"key": "value"}"#.to_string())),
+ test_message(Payload::Proto("[1, 2]".to_string())),
+ ];
+
+ let extracted = sink.extract_json_payloads(messages);
+
+ assert_eq!(
+ extracted,
+ vec![
+ simd_json::json!({"key": "value"}),
+ simd_json::json!({"data": [1, 2], "data_type": "json"}),
+ ]
+ );
+ }
+
#[test]
fn given_raw_nonobjects_when_extracted_should_preserve_original_bytes() {
let sink = QuickwitSink::new(1, test_config());
diff --git a/core/connectors/sinks/s3_sink/README.md b/core/connectors/sinks/s3_sink/README.md
index c3bf0c1aa..c988c3f45 100644
--- a/core/connectors/sinks/s3_sink/README.md
+++ b/core/connectors/sinks/s3_sink/README.md
@@ -129,7 +129,7 @@ Each line:
{"offset":42,"timestamp":"2026-03-16T14:02:31Z","stream":"application_logs","topic":"api_requests","partition_id":1,"payload":{"method":"GET","path":"/api/users","status":200}}
```
-Both JSON formats wrap the payload under `payload`, including when metadata is disabled. JSON values retain their shape; text and Proto variants become strings. Valid JSON raw bytes are parsed; other raw bytes, Avro and FlatBuffer variants become base64 strings without an encoding tag.
+Both JSON formats wrap the payload under `payload`, including when metadata is disabled. JSON values retain their shape; text variants become strings, and Proto variants become the JSON document they hold when the text parses as JSON (the descriptor-less `proto_convert` fallback) or a string otherwise. Valid JSON raw bytes are parsed; other raw bytes, Avro and FlatBuffer variants become base64 strings without an encoding tag.
Metadata timestamps use UTC second precision; IDs, checksums and origin timestamps are omitted. Headers are independent of metadata: strings remain strings, raw values become base64, booleans remain booleans, supported numeric values become JSON numbers, and other values use strings.
Raw output concatenates payload bytes without delimiters and ignores both inclusion flags. Use `schema = "raw"` without payload-changing transforms to preserve original bytes; JSON decoding followed by raw output can reserialize the payload.
diff --git a/core/connectors/sinks/s3_sink/src/formatter.rs b/core/connectors/sinks/s3_sink/src/formatter.rs
index 05b2b08d3..a70ecab46 100644
--- a/core/connectors/sinks/s3_sink/src/formatter.rs
+++ b/core/connectors/sinks/s3_sink/src/formatter.rs
@@ -173,7 +173,13 @@ fn payload_to_json_value(payload: &Payload) -> Value {
Ok(v) => v,
Err(_) => Value::String(base64_encode(bytes)),
},
- Payload::Proto(text) => Value::String(text.clone()),
+ // Proto text holding JSON is the descriptor-less `proto_convert`
+ // fallback and is written as the document it holds, the way the same
+ // bytes were written when the batch was tagged `json`.
+ Payload::Proto(text) => match payload.json_document() {
+ Some(document) => owned_value_to_serde_json(document.as_ref()),
+ None => Value::String(text.clone()),
+ },
Payload::FlatBuffer(bytes) => Value::String(base64_encode(bytes)),
Payload::Avro(bytes) => Value::String(base64_encode(bytes)),
}
@@ -319,6 +325,26 @@ mod tests {
assert_eq!(value["headers"]["content-type"], "application/json");
}
+ #[test]
+ fn proto_payload_holding_json_is_written_as_a_document() {
+ let payload = Payload::Proto(r#"{"id":1,"name":"row-1"}"#.to_string());
+
+ assert_eq!(
+ payload_to_json_value(&payload),
+ serde_json::json!({"id": 1, "name": "row-1"})
+ );
+ }
+
+ #[test]
+ fn proto_text_that_is_not_json_is_written_as_a_string() {
+ let payload = Payload::Proto("name: \"row-1\"".to_string());
+
+ assert_eq!(
+ payload_to_json_value(&payload),
+ Value::String("name: \"row-1\"".to_string())
+ );
+ }
+
#[test]
fn raw_format() {
let payload = Payload::Text("hello world".to_string());
diff --git a/core/connectors/sinks/surrealdb_sink/README.md b/core/connectors/sinks/surrealdb_sink/README.md
index 4a57b93cc..8a3f7e2cf 100644
--- a/core/connectors/sinks/surrealdb_sink/README.md
+++ b/core/connectors/sinks/surrealdb_sink/README.md
@@ -120,7 +120,12 @@ Every submitted record contains:
- `payload_encoding`
`include_metadata` adds `iggy_stream`, `iggy_topic`, `iggy_partition_id`,
-`iggy_offset`, `iggy_timestamp` and `iggy_schema`. The checksum, origin timestamp
+`iggy_offset`, `iggy_timestamp` and `iggy_schema`. `iggy_schema` records the
+payload variant the connector received, not the stream's configured `schema`:
+an `avro` stream writes `json`, because the Avro decoder extracts to JSON by
+default, and `flat_buffer` and `proto` streams do the same. Rows written before
+this change recorded the configured schema, so historical rows on those streams
+disagree with new ones. The checksum, origin timestamp
and headers have independent inclusion flags. `iggy_headers` is omitted when
headers are absent or empty. Non-raw headers are strings, including numeric and
boolean values; raw headers use `{"data":"AQID","iggy_header_encoding":"base64"}`.
@@ -131,7 +136,9 @@ then partition, offset and the 32-digit hexadecimal message ID. Payload and
transform changes do not alter that identity.
`payload_format = "auto"` stores decoded JSON payloads as queryable SurrealDB
-values, text/Proto variants as strings, and raw/Avro/FlatBuffer bytes as base64
+values, text variants as strings, Proto variants as the JSON document they
+hold when the text parses as JSON or as strings otherwise, and
+raw/Avro/FlatBuffer bytes as base64
strings, even when raw bytes contain valid JSON. Explicit `json` parses other
payload variants as JSON, `text` requires UTF-8, and `base64`/`binary` encodes the
payload bytes. Invalid conversions reject that record. Destination schema,
diff --git a/core/connectors/sinks/surrealdb_sink/src/lib.rs b/core/connectors/sinks/surrealdb_sink/src/lib.rs
index b697ceda1..994d4373b 100644
--- a/core/connectors/sinks/surrealdb_sink/src/lib.rs
+++ b/core/connectors/sinks/surrealdb_sink/src/lib.rs
@@ -893,6 +893,7 @@ fn build_insert_query(table: &str, records: &[Value]) -> Result<Bytes, Error> {
}
fn build_auto_payload_document(payload: Payload) -> Result<PayloadDocument, Error> {
+ let payload = payload.into_json_document();
match payload {
Payload::Json(value) => Ok(PayloadDocument {
value: owned_value_to_serde_json(&value),
@@ -1611,6 +1612,22 @@ mod tests {
assert_eq!(document.value, json!({"name": "Alice", "active": true}));
}
+ #[test]
+ fn given_auto_payload_proto_holding_json_should_store_queryable_json() {
+ let payload = Payload::Proto(r#"{"id":1,"name":"row-1"}"#.to_string());
+ let document = build_auto_payload_document(payload).expect("Failed to build payload");
+ assert_eq!(document.value, json!({"id": 1, "name": "row-1"}));
+ assert_eq!(document.encoding, ENCODING_JSON);
+ }
+
+ #[test]
+ fn given_auto_payload_proto_text_should_store_text() {
+ let payload = Payload::Proto("name: \"row-1\"".to_string());
+ let document = build_auto_payload_document(payload).expect("Failed to build payload");
+ assert_eq!(document.value, Value::String("name: \"row-1\"".to_string()));
+ assert_eq!(document.encoding, ENCODING_TEXT);
+ }
+
#[test]
fn given_auto_payload_text_should_store_text() {
let payload = Payload::Text("hello".to_string());
diff --git a/core/connectors/sources/influxdb_source/dependencies.md b/core/connectors/sources/influxdb_source/dependencies.md
index d31cf1a29..84d0ed0e9 100644
--- a/core/connectors/sources/influxdb_source/dependencies.md
+++ b/core/connectors/sources/influxdb_source/dependencies.md
@@ -43,7 +43,7 @@ to `cargo tree -p iggy_connector_influxdb_source` for the full graph.
| `csv` | `^1.4.0` | MIT / Apache-2.0 | Parses InfluxDB V2 annotated-CSV query responses in `row.rs::parse_csv_rows`. |
| `dashmap` | `^6.1.0` | MIT | Concurrent hash map; injected into this crate's namespace by the `source_connector!` macro expansion in the SDK. Not used directly in source files. |
| `iggy_common` | `^0.11.0` | Apache-2.0 | Shared Iggy types: `DateTime`, `Utc`, and `serde_secret` used for safe token serialisation in config structs. |
-| `iggy_connector_sdk` | `^0.4.0` | Apache-2.0 | Core connector abstractions: `Source` trait, `ProducedMessage`, `ProducedMessages`, `ConnectorState`, `Schema`, `Error`, retry/circuit-breaker utilities, and the `source_connector!` registration macro. |
+| `iggy_connector_sdk` | `^0.5.0` | Apache-2.0 | Core connector abstractions: `Source` trait, `ProducedMessage`, `ProducedMessages`, `ConnectorState`, `Schema`, `Error`, retry/circuit-breaker utilities, and the `source_connector!` registration macro. |
| `regex` | `^1.12.3` | MIT / Apache-2.0 | Compiles the RFC 3339 cursor-validation regex once via `OnceLock` in `common.rs::cursor_re()`. |
| `reqwest` | `^0.13.3` | MIT / Apache-2.0 | Async HTTP client used to issue Flux (V2) and SQL (V3) query requests to InfluxDB. |
| `reqwest-middleware` | `^0.5.1` | MIT | Middleware wrapper around `reqwest::Client` that attaches the retry and tracing layers built by `iggy_connector_sdk::retry::build_retry_client`. |
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index eb63750a3..ba6ff18a4 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -33,6 +33,7 @@ default = ["login-session"]
login-session = ["dep:zbus-secret-service-keyring-store"]
[dependencies]
+apache-avro = { workspace = true }
arrow = { workspace = true }
assert_cmd = { workspace = true }
async-trait = { workspace = true }
diff --git a/core/integration/tests/connectors/clickhouse/mod.rs b/core/integration/tests/connectors/clickhouse/mod.rs
index 43c559e55..4b34987ae 100644
--- a/core/integration/tests/connectors/clickhouse/mod.rs
+++ b/core/integration/tests/connectors/clickhouse/mod.rs
@@ -18,3 +18,5 @@
mod clickhouse_sink;
const TEST_MESSAGE_COUNT: usize = 3;
+
+mod proto_text;
diff --git a/core/integration/tests/connectors/clickhouse/proto_text.rs b/core/integration/tests/connectors/clickhouse/proto_text.rs
new file mode 100644
index 000000000..68dfe9e7a
--- /dev/null
+++ b/core/integration/tests/connectors/clickhouse/proto_text.rs
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::TEST_MESSAGE_COUNT;
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::ClickHouseSinkFixture;
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use integration::harness::seeds;
+use integration::iggy_harness;
+
+/// A `proto_convert` transform with no descriptor falls back to proto text, so
+/// every message reaches the sink as `Payload::Proto` holding the JSON it was
+/// given. The default JSONEachRow builder reads that as the document it holds.
+/// A builder that only accepts `Payload::Json` skips every row, sends nothing,
+/// and returns success, so the offset is committed and the rows are gone; this
+/// test times out on that with zero rows and zero errors counted.
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/clickhouse/proto_text.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_a_proto_convert_transform_when_the_sink_consumes_should_store_the_rows(
+ harness: &TestHarness,
+ fixture: ClickHouseSinkFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let messages_data = create_test_messages(TEST_MESSAGE_COUNT);
+ let mut messages: Vec<IggyMessage> = messages_data
+ .iter()
+ .enumerate()
+ .map(|(i, msg)| {
+ let payload = serde_json::to_vec(msg).expect("Failed to serialize message");
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(Bytes::from(payload))
+ .build()
+ .expect("Failed to build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ fixture
+ .wait_for_rows(TEST_MESSAGE_COUNT)
+ .await
+ .expect("the proto text batch must be inserted, not skipped");
+
+ let rows = fixture
+ .fetch_rows()
+ .await
+ .expect("Failed to fetch rows from ClickHouse");
+
+ assert_eq!(
+ rows.len(),
+ TEST_MESSAGE_COUNT,
+ "Expected {TEST_MESSAGE_COUNT} rows in ClickHouse"
+ );
+
+ for (i, row) in rows.iter().enumerate() {
+ let expected = &messages_data[i];
+
+ let id = row["id"]
+ .as_str()
+ .and_then(|s| s.parse::<u64>().ok())
+ .or_else(|| row["id"].as_u64())
+ .unwrap_or_else(|| panic!("Missing 'id' at row {i}"));
+ assert_eq!(id, expected.id, "id mismatch at row {i}");
+
+ let name = row["name"]
+ .as_str()
+ .unwrap_or_else(|| panic!("Missing 'name' at row {i}"));
+ assert_eq!(name, expected.name, "name mismatch at row {i}");
+
+ let amount = row["amount"]
+ .as_str()
+ .and_then(|s| s.parse::<f64>().ok())
+ .or_else(|| row["amount"].as_f64())
+ .unwrap_or_else(|| panic!("Missing 'amount' at row {i}"));
+ assert!(
+ (amount - expected.amount).abs() < 1e-6,
+ "amount mismatch at row {i}: got {amount}, expected {}",
+ expected.amount
+ );
+ }
+}
diff --git a/core/integration/tests/connectors/clickhouse/proto_text.toml b/core/integration/tests/connectors/clickhouse/proto_text.toml
new file mode 100644
index 000000000..e80913604
--- /dev/null
+++ b/core/integration/tests/connectors/clickhouse/proto_text.toml
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[connectors]
+config_type = "local"
+config_dir = "tests/connectors/clickhouse/proto_text_config"
diff --git a/core/integration/tests/connectors/clickhouse/proto_text_config/clickhouse_sink.toml b/core/integration/tests/connectors/clickhouse/proto_text_config/clickhouse_sink.toml
new file mode 100644
index 000000000..107820e11
--- /dev/null
+++ b/core/integration/tests/connectors/clickhouse/proto_text_config/clickhouse_sink.toml
@@ -0,0 +1,67 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# `ClickHouseSinkFixture::connectors_runtime_envs` overrides the plugin path,
+# the connection settings and every `[[streams]]` key at run time, so those
+# values are structural rather than meaningful. The `[transforms.proto_convert]`
+# block below is the only part of this file specific to the test, and
+# `insert_format` is left at the default so the JSONEachRow builder is the one
+# handed the proto text.
+
+type = "sink"
+key = "clickhouse"
+enabled = true
+version = 0
+name = "ClickHouse sink"
+path = "../../target/debug/libiggy_connector_clickhouse_sink"
+verbose = false
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 1000
+poll_interval = "5ms"
+consumer_group = "clickhouse_sink_connector"
+
+# No descriptor is configured, so `json_to_protobuf` falls back to proto text
+# and every message reaches the sink as `Payload::Proto` holding its JSON.
+[transforms.proto_convert]
+enabled = true
+source_format = "json"
+target_format = "proto"
+include_paths = []
+preserve_unknown_fields = false
+
+[transforms.proto_convert.conversion_options]
+validate_messages = true
+pretty_json = false
+include_metadata = false
+type_url_prefix = "type.googleapis.com"
+strict_mode = false
+
+[plugin_config]
+url = "http://localhost:8123"
+database = "default"
+username = "default"
+password = ""
+table = "events"
+insert_format = "json_each_row"
+timeout_seconds = 30
+max_retries = 3
+retry_delay = 1
+verbose_logging = false
diff --git a/core/integration/tests/connectors/doris/mod.rs b/core/integration/tests/connectors/doris/mod.rs
index 7b1034d81..66f5572ad 100644
--- a/core/integration/tests/connectors/doris/mod.rs
+++ b/core/integration/tests/connectors/doris/mod.rs
@@ -17,3 +17,4 @@
mod doris_sink;
mod doris_stream_load_format_bench;
+mod proto_text;
diff --git a/core/integration/tests/connectors/doris/proto_text.rs b/core/integration/tests/connectors/doris/proto_text.rs
new file mode 100644
index 000000000..121ced7d5
--- /dev/null
+++ b/core/integration/tests/connectors/doris/proto_text.rs
@@ -0,0 +1,77 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{DorisOps, DorisSinkPreCreatedFixture};
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use integration::harness::seeds;
+use integration::iggy_harness;
+
+const TEST_TABLE: &str = "test_topic";
+
+/// A `proto_convert` transform with no descriptor falls back to proto text, so
+/// every message reaches the sink as `Payload::Proto` holding the JSON it was
+/// given. The sink loads that as the document it holds. A sink that only accepts
+/// `Payload::Json` aborts the whole poll on the first message, so no chunk is
+/// written and the committed offset is not replayed; this test times out on
+/// that with zero rows.
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/doris/proto_text.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_proto_text_messages_should_store(
+ harness: &TestHarness,
+ fixture: DorisSinkPreCreatedFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let message_count = 10;
+ let test_messages = create_test_messages(message_count);
+ let mut messages: Vec<IggyMessage> = test_messages
+ .iter()
+ .enumerate()
+ .map(|(i, m)| {
+ let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(payload)
+ .build()
+ .expect("build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ let count = fixture
+ .wait_for_rows(fixture.database(), TEST_TABLE, message_count as i64)
+ .await
+ .expect("the proto text batch must be loaded, not aborted");
+ assert_eq!(count, message_count as i64);
+}
diff --git a/core/integration/tests/connectors/doris/proto_text.toml b/core/integration/tests/connectors/doris/proto_text.toml
new file mode 100644
index 000000000..ba4527964
--- /dev/null
+++ b/core/integration/tests/connectors/doris/proto_text.toml
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[connectors]
+config_type = "local"
+config_dir = "tests/connectors/doris/proto_text_config"
diff --git a/core/integration/tests/connectors/doris/proto_text_config/doris_sink.toml b/core/integration/tests/connectors/doris/proto_text_config/doris_sink.toml
new file mode 100644
index 000000000..f6b8ff982
--- /dev/null
+++ b/core/integration/tests/connectors/doris/proto_text_config/doris_sink.toml
@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# `DorisSinkPreCreatedFixture::connectors_runtime_envs` overrides the plugin
+# path, the FE URL, the database and every `[[streams]]` key at run time, so
+# those values are structural rather than meaningful. The
+# `[transforms.proto_convert]` block below is the only part of this file
+# specific to the test.
+
+type = "sink"
+key = "doris"
+enabled = true
+version = 0
+name = "Doris sink"
+path = "../../target/debug/libiggy_connector_doris_sink"
+verbose = false
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "doris_sink"
+
+# No descriptor is configured, so `json_to_protobuf` falls back to proto text
+# and every message reaches the sink as `Payload::Proto` holding its JSON.
+[transforms.proto_convert]
+enabled = true
+source_format = "json"
+target_format = "proto"
+include_paths = []
+preserve_unknown_fields = false
+
+[transforms.proto_convert.conversion_options]
+validate_messages = true
+pretty_json = false
+include_metadata = false
+type_url_prefix = "type.googleapis.com"
+strict_mode = false
+
+[plugin_config]
+fe_url = "http://localhost:8030"
+database = ""
+table = ""
+username = "root"
+password = ""
+label_prefix = "iggy"
+batch_size = 1000
+timeout = "30s"
diff --git a/core/integration/tests/connectors/elasticsearch/mod.rs b/core/integration/tests/connectors/elasticsearch/mod.rs
index 61d6b86c2..0106e6378 100644
--- a/core/integration/tests/connectors/elasticsearch/mod.rs
+++ b/core/integration/tests/connectors/elasticsearch/mod.rs
@@ -17,6 +17,7 @@
mod elasticsearch_sink;
mod elasticsearch_source;
+mod proto_text;
const TEST_MESSAGE_COUNT: usize = 3;
const POLL_ATTEMPTS: usize = 100;
diff --git a/core/integration/tests/connectors/elasticsearch/proto_text.rs b/core/integration/tests/connectors/elasticsearch/proto_text.rs
new file mode 100644
index 000000000..bd121045c
--- /dev/null
+++ b/core/integration/tests/connectors/elasticsearch/proto_text.rs
@@ -0,0 +1,126 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::TEST_MESSAGE_COUNT;
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::ElasticsearchSinkFixture;
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use integration::harness::seeds;
+use integration::iggy_harness;
+
+/// A `proto_convert` transform with no descriptor falls back to proto text, so
+/// the batch is tagged `Schema::Proto` and the sink is handed `Payload::Proto`
+/// holding the JSON it was given. The sink indexes that as the document it
+/// holds, field by field, the same way the bytes indexed when the batch was
+/// tagged `json`. Indexing it as one opaque `text` field would pass while
+/// losing every original field, so the assertions read the fields back.
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/elasticsearch/proto_text.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_a_proto_convert_transform_when_the_sink_consumes_should_index_the_document(
+ harness: &TestHarness,
+ fixture: ElasticsearchSinkFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let messages_data = create_test_messages(TEST_MESSAGE_COUNT);
+ let mut messages: Vec<IggyMessage> = messages_data
+ .iter()
+ .enumerate()
+ .map(|(i, msg)| {
+ let payload = serde_json::to_vec(msg).expect("Failed to serialize message");
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(Bytes::from(payload))
+ .build()
+ .expect("Failed to build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ fixture
+ .wait_for_documents(TEST_MESSAGE_COUNT)
+ .await
+ .expect("the proto text batch must be indexed, not dropped");
+
+ fixture
+ .refresh_index()
+ .await
+ .expect("Failed to refresh index");
+
+ let search_result = fixture
+ .search_documents()
+ .await
+ .expect("Failed to search documents");
+
+ assert_eq!(
+ search_result.hits.total.value, TEST_MESSAGE_COUNT,
+ "Expected {TEST_MESSAGE_COUNT} documents in Elasticsearch"
+ );
+
+ for (i, hit) in search_result.hits.hits.iter().enumerate() {
+ assert!(
+ hit.source.get("data_type").is_none(),
+ "a proto payload holding JSON must index as a document, not a text blob: {}",
+ hit.source
+ );
+ let name = hit
+ .source
+ .get("name")
+ .and_then(|value| value.as_str())
+ .unwrap_or_else(|| {
+ panic!(
+ "the original fields must survive the transform: {}",
+ hit.source
+ )
+ });
+ assert!(
+ name.starts_with("user_"),
+ "unexpected name at hit {i}: {}",
+ hit.source
+ );
+ assert!(
+ hit.source
+ .get("amount")
+ .and_then(|value| value.as_f64())
+ .is_some(),
+ "amount must be indexed as a number, got {}",
+ hit.source
+ );
+ assert!(
+ hit.source.get("_iggy_offset").is_some(),
+ "metadata must still be injected into the document, got {}",
+ hit.source
+ );
+ }
+}
diff --git a/core/integration/tests/connectors/elasticsearch/proto_text.toml b/core/integration/tests/connectors/elasticsearch/proto_text.toml
new file mode 100644
index 000000000..061fe4947
--- /dev/null
+++ b/core/integration/tests/connectors/elasticsearch/proto_text.toml
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[connectors]
+config_type = "local"
+config_dir = "tests/connectors/elasticsearch/proto_text_config"
diff --git a/core/integration/tests/connectors/elasticsearch/proto_text_config/elasticsearch_sink.toml b/core/integration/tests/connectors/elasticsearch/proto_text_config/elasticsearch_sink.toml
new file mode 100644
index 000000000..e9fa1945f
--- /dev/null
+++ b/core/integration/tests/connectors/elasticsearch/proto_text_config/elasticsearch_sink.toml
@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# `ElasticsearchSinkFixture::connectors_runtime_envs` overrides the plugin path,
+# the URL, the index and every `[[streams]]` key at run time, so those values are
+# structural rather than meaningful. The `[transforms.proto_convert]` block below
+# is the only part of this file specific to the test.
+
+type = "sink"
+key = "elasticsearch"
+enabled = true
+version = 0
+name = "Elasticsearch sink"
+path = "../../target/debug/libiggy_connector_elasticsearch_sink"
+verbose = false
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "elasticsearch_sink_cg"
+
+# No descriptor is configured, so `json_to_protobuf` falls back to proto text
+# and every message reaches the sink as `Payload::Proto`.
+[transforms.proto_convert]
+enabled = true
+source_format = "json"
+target_format = "proto"
+include_paths = []
+preserve_unknown_fields = false
+
+[transforms.proto_convert.conversion_options]
+validate_messages = true
+pretty_json = false
+include_metadata = false
+type_url_prefix = "type.googleapis.com"
+strict_mode = false
+
+[plugin_config]
+url = "http://localhost:9200"
+index = "iggy_messages"
+create_index_if_not_exists = true
+timeout_seconds = 30
+max_retries = 3
+retry_delay = "1s"
+retry_max_delay = "5s"
diff --git a/core/integration/tests/connectors/runtime/mod.rs b/core/integration/tests/connectors/runtime/mod.rs
index 72f3ed271..3c74f48d7 100644
--- a/core/integration/tests/connectors/runtime/mod.rs
+++ b/core/integration/tests/connectors/runtime/mod.rs
@@ -18,3 +18,4 @@
mod benchmark;
mod error_isolation;
mod http_state;
+mod schema_tagging;
diff --git a/core/integration/tests/connectors/runtime/schema_tagging.rs b/core/integration/tests/connectors/runtime/schema_tagging.rs
new file mode 100644
index 000000000..c9fca9265
--- /dev/null
+++ b/core/integration/tests/connectors/runtime/schema_tagging.rs
@@ -0,0 +1,134 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use apache_avro::Schema as AvroSchema;
+use apache_avro::types::Value as AvroValue;
+use apache_avro::writer::datum::GenericDatumWriter;
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use integration::harness::{TestHarness, seeds};
+use integration::iggy_harness;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// The runtime tags each batch with a `Schema` and the SDK rebuilds the sink's
+// `Payload` from that tag alone. The Avro decoder returns JSON, so the tag has
+// to say `json` even though the stream is configured `avro`. Asserting on the
+// stdout sink's own log lines is the cheapest way to see what actually crossed
+// the FFI boundary, and it needs no container.
+
+const MESSAGE_COUNT: usize = 5;
+const SCHEMA: &str = r#"{"type":"record","name":"Event","fields":[{"name":"id","type":"long"},{"name":"name","type":"string"}]}"#;
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/runtime/schema_tagging.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_an_avro_stream_when_the_sink_consumes_should_receive_json_payloads(
+ harness: &TestHarness,
+) {
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let schema = AvroSchema::parse_str(SCHEMA).expect("failed to parse the Avro schema");
+ let mut messages: Vec<IggyMessage> = (0..MESSAGE_COUNT)
+ .map(|index| {
+ IggyMessage::builder()
+ .id((index + 1) as u128)
+ .payload(Bytes::from(avro_datum(&schema, index as i64)))
+ .build()
+ .expect("failed to build the message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("failed to send the messages");
+
+ let logs = wait_for_sink_batch(harness).await;
+
+ assert!(
+ logs.contains("schema: json,"),
+ "the sink should be told the batch is JSON, got:\n{logs}"
+ );
+ assert!(
+ !logs.contains("schema: avro,"),
+ "the batch must not be tagged with the decoder's schema, got:\n{logs}"
+ );
+ assert!(
+ !logs.contains("Avro("),
+ "the sink must not receive a Payload::Avro, got:\n{logs}"
+ );
+ assert!(
+ logs.contains("Json("),
+ "the sink should receive a Payload::Json, got:\n{logs}"
+ );
+ for index in 0..MESSAGE_COUNT {
+ let name = format!("row-{index}");
+ assert!(
+ logs.contains(&name),
+ "decoded payload should carry '{name}', got:\n{logs}"
+ );
+ }
+}
+
+/// Bare Avro datum, not an object container file: the decoder reads with
+/// `GenericDatumReader::read_value` and rejects trailing bytes.
+fn avro_datum(schema: &AvroSchema, id: i64) -> Vec<u8> {
+ let record = AvroValue::Record(vec![
+ ("id".to_owned(), AvroValue::Long(id)),
+ ("name".to_owned(), AvroValue::String(format!("row-{id}"))),
+ ]);
+ GenericDatumWriter::builder(schema)
+ .build()
+ .expect("failed to build the Avro writer")
+ .write_value_to_vec(record)
+ .expect("failed to encode the Avro datum")
+}
+
+async fn wait_for_sink_batch(harness: &TestHarness) -> String {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available");
+ let deadline = Instant::now() + Duration::from_secs(15);
+ let mut logs = String::new();
+
+ // The per-message payload lines follow the batch header, and the log file
+ // is read while it is still being written, so waiting on the header alone
+ // can return before the payloads the assertions read.
+ while Instant::now() < deadline {
+ let (stdout, stderr) = runtime.collect_logs();
+ logs = format!("{stdout}\n{stderr}");
+ if logs.contains("Stdout sink with ID:")
+ && logs.matches("Message offset:").count() >= MESSAGE_COUNT
+ {
+ return logs;
+ }
+ sleep(Duration::from_millis(200)).await;
+ }
+
+ panic!("the stdout sink never reported {MESSAGE_COUNT} messages. logs:\n{logs}");
+}
diff --git a/core/integration/tests/connectors/runtime/schema_tagging.toml b/core/integration/tests/connectors/runtime/schema_tagging.toml
new file mode 100644
index 000000000..7bf12180a
--- /dev/null
+++ b/core/integration/tests/connectors/runtime/schema_tagging.toml
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[connectors]
+config_type = "local"
+config_dir = "tests/connectors/runtime/schema_tagging_config"
diff --git a/core/integration/tests/connectors/runtime/schema_tagging_config/stdout_sink.toml b/core/integration/tests/connectors/runtime/schema_tagging_config/stdout_sink.toml
new file mode 100644
index 000000000..3e2162012
--- /dev/null
+++ b/core/integration/tests/connectors/runtime/schema_tagging_config/stdout_sink.toml
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "sink"
+key = "stdout_schema"
+enabled = true
+version = 0
+name = "Stdout schema tagging sink"
+path = "../../target/debug/libiggy_connector_stdout_sink"
+verbose = true
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "avro"
+avro_schema_json = '{"type":"record","name":"Event","fields":[{"name":"id","type":"long"},{"name":"name","type":"string"}]}'
+batch_length = 10
+poll_interval = "5ms"
+consumer_group = "schema_tagging_sink_connector"
+
+[plugin_config]
+print_payload = true