Kafka Connect: Preserve explicit null values instead of replacing them with schema defaults

[moggaa]

Closes #17652

Problem

Kafka Connect's Struct.get(Field) returns the field's schema defaultValue when the stored value is null, without checking isOptional() (KAFKA-8713). The sink and its bundled SMTs read field values with get(), so an explicit NULL in the source is silently written to the Iceberg table as the column default — no error or warning is raised.

The most common trigger is Debezium CDC, which propagates a column's DDL DEFAULT clause into the Connect schema's defaultValue: for any column that is nullable with a non-NULL default (e.g. MySQL VARCHAR(255) NULL DEFAULT ''), every explicitly NULL value is replaced:

source row: NULL  →  Kafka message: null (correct)  →  Iceberg table: '' (corrupted)

We hit this in production: for affected columns, 100% of NULLs had been written as the column default. No converter configuration can prevent it — even when the deserializing converter preserves the null, the Connect schema still carries the defaultValue, and the sink's own reads re-apply it on every Struct.get().

The same bug class has been fixed across the ecosystem: Debezium uses getWithoutDefault unconditionally in its own SMTs (ExtractNewRecordState#L191-L193), Kafka core added a replace.null.with.default option to nine SMTs (KIP-1040) and to JsonConverter (KIP-581), and the JDBC sinks fixed it as well (confluentinc/kafka-connect-jdbc#1433, DBZ-7191).

Changes

A Struct preserves an explicit null only as long as nobody copies it with get(): the copied schema still carries the defaultValue, so the first get() in the chain irreversibly bakes the default into the stored value. The SMTs are therefore pure pass-throughs, and the single behavioral decision happens at the final read, controlled by one sink option:

• Bundled SMTs (DebeziumTransform, KafkaMetadataTransform, CopyValue): copy fields with getWithoutDefault so the stored null survives the copy, as Debezium's own SMTs do. This is not an observable behavior change on the default path: the output schema still carries the defaultValue, so any consumer reading with get() (including today's RecordConverter) receives exactly the same values as before. The new SMT tests assert both halves (the stored null and the retained schema default).
• IcebergSinkConfig: new option iceberg.tables.replace-null-with-default, default true, which preserves the current behavior. Set to false to keep explicit nulls.
• RecordConverter: struct field reads, including the variant conversion path, go through a shared helper gated by the option.
• RecordUtils / SinkWriter: route-field extraction is gated by the same option: with the option disabled, an explicitly null route field is skipped like any other null route value, keeping writes and routing consistent.
• Docs: config table row plus a note on the routing behavior and on the JSON converter's own replace.null.with.default setting.

Design notes

• Why unconditional SMT changes instead of per-SMT options (the KIP-1040 pattern): a single SMT in the chain left at true silently defeats the sink-level option, turning one bug into a configuration puzzle; and as noted above, the unconditional change is not observable to get() readers. Happy to switch to per-SMT options if that's preferred; the sink-level option composes with either choice.
• Why the default is true: purely for backward compatibility. Given that a null that reaches the sink is always an explicitly written null, there is also a case for defaulting to false; happy to go either way.

Testing

Unit tests cover the transform, converter (struct and variant), routing, and config-default paths; each new behavior is parameterized over both option values. All kafka-connect module tests pass.

---

Generative AI (Claude) was used to help draft the changes and tests.
All changes were reviewed and verified by the author.

[uros-b]

cc @waterWang here regarding #17684 (review).