Native Kafka sink: Rust batch serialization with Flink exactly-once transactions

[jordepic]

What

Build a columnar Kafka sink path that keeps Flink's KafkaSink writer, committer, transaction naming, checkpoint state, recovery, and broker IO unchanged, but serializes each incoming Arrow batch in Rust. The JVM receives already-serialized key/value bytes and hands them to the stock KafkaSink. This is the agreed hybrid: native format work, Flink-owned exactly-once transaction lifecycle.

Contract

• The native operator consumes ArrowBatch and emits pre-serialized Kafka records.
• Serialization crosses JNI once per Arrow batch; Java materializes the byte arrays required by KafkaProducer's Serializer contract. There is no RowData transpose or Java format encoder.
• KafkaSink remains authoritative for NONE, AT_LEAST_ONCE, and EXACTLY_ONCE delivery, transactional IDs, writer state, recovery abort/commit, metrics, partition lookup, and producer configuration.
• Unsupported formats, key/value projections, writable metadata, partitioners, or upsert buffering fall back to the stock table sink rather than silently diverging.
• Native output must be byte-for-byte parity tested against Flink's format serializer for every admitted type and option.

Implementation slices

• Matcher/physical/exec-node skeleton for singleton-topic kafka tables, preserving stock fallback.
• Batched native JSON value serializer with Flink-exact scalar, decimal, binary, date/time/timestamp, null-field, and map-null-key semantics.
• Optional native key serialization and key/value projections.
• Batch drain ABI and Java pre-serialized record type; document the unavoidable native-to-byte[] copy required by KafkaProducer.
• Feed records into stock KafkaSink with producer properties, delivery guarantee, transaction naming strategy, transactional-id prefix, and sink parallelism preserved.
• Changelog/upsert semantics, tombstones, writable metadata (topic, timestamp, headers), and custom partitioner parity or explicit fallback.
• CSV, Avro, and protobuf encoders after JSON; each parity-gated independently.
• Native memory accounting and cancellation/close behavior.
• Metrics separating native serialization time/bytes from Kafka writer/send/checkpoint time.

Correctness acceptance

• Byte-for-byte native-vs-Flink serializer corpus, including null/nested/temporal/decimal edges and all admitted options.
• NONE and AT_LEAST_ONCE broker integration tests.
• EXACTLY_ONCE MiniCluster failover test: completed and in-flight checkpoints, task failure/restart, no visible aborted duplicates through a read_committed verifier.
• Rescale/restore and transactional-id stability test.
• Native Kafka source -> native operators -> native serialization -> stock KafkaSink test with no RowData transpose.
• Fallback tests for every unsupported sink shape.

Performance acceptance

• Criterion benchmark native batch serialization versus a non-native equivalent corpus, with rows/s, bytes/s, allocations, and batch-size sweep.
• Profile serialization and JNI byte materialization; optimize the batch drain before changing the correctness boundary.
• Nexmark exactly-once matrix using the same Kafka cluster for input and output: stock Flink vs StreamFusion, mini-batch disabled and enabled.
• Update README headline chart from blackhole output to exactly-once Kafka output, reporting throughput plus checkpoint duration/recovery replay cost.

References

• Flink KafkaDynamicSink, DynamicKafkaRecordSerializationSchema, KafkaSinkWriter, KafkaCommitter, and KafkaWriterStateSerializer are the semantic authority.
• rust-rdkafka is a serialization/representation reference only for this hybrid; it does not own transaction recovery.
• RisingWave and Proton remain architecture/performance references, not semantic authorities.

[jordepic]

Milestone shipped in 64205f4: native upsert-kafka now batch-serializes default primary-key JSON keys and ALL-field JSON values, maps UPDATE_BEFORE/DELETE to tombstones, and keeps Flink KafkaSink in charge of exactly-once transactions. A broker-backed updating aggregate test verifies committed final state; sink-schema field names are pinned so planner-generated names cannot leak onto the wire. Explicit key/value projections, metadata/partitioners/buffering, and restore-rescale coverage remain open.