moggaa opened a new issue, #17652: URL: https://github.com/apache/iceberg/issues/17652
### Apache Iceberg version main (development) ### Query engine Kafka Connect ### Please describe the bug 🐞 ### Problem Kafka Connect's `Struct.get(Field)` returns the field's schema `defaultValue` when the stored value is `null`, without checking `isOptional()` (longstanding Connect behavior, see [KAFKA-8713](https://issues.apache.org/jira/browse/KAFKA-8713)). The Iceberg sink and its bundled SMTs read field values with `get()` in several places, so an explicit `NULL` in the source ends up written to the Iceberg table as the column's default value. No error or warning is raised; the data is simply wrong. Affected call sites (permalinks to current `main`, 35889387): - `DebeziumTransform` — copies the Debezium `before`/`after` payload into the new value: [DebeziumTransform.java#L110](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/DebeziumTransform.java#L110) - `KafkaMetadataTransform` — copies top-level fields into the new value: [KafkaMetadataTransform.java#L246](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/KafkaMetadataTransform.java#L246) - `CopyValue` — struct copy loop and source-field copy: [CopyValue.java#L93-L95](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/CopyValue.java#L93-L95) - `RecordConverter` — struct field reads, including the variant conversion path: [RecordConverter.java#L262](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java#L262), [#L631](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java#L631), [#L666](https://github.com/apache/iceberg/blob/35889387c8c6ff0a3f57d17a304709dc1b7d9340/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java#L666) The behavior has been present since these components were first introduced (the record converter in #9641, the Debezium SMT in #11936). ### When this is reached The most common trigger is Debezium CDC. Debezium propagates a column's DDL `DEFAULT` clause into the Connect schema's `defaultValue`. So 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. The Kafka messages were correct; the corruption happens entirely inside the sink. Two things make this impossible to work around with configuration: 1. **No converter configuration can prevent it.** We tested the available converter knobs on the sink side — `replace.null.with.default=false` on the JSON converter and `ignore.default.for.nullables=true` on Confluent's Avro converter (Confluent Platform 7.5.3) — and the table still receives the default. This is expected from the code: even when the deserializing converter preserves the null, the Connect schema still carries the `defaultValue`, and the sink's own reads (the SMT copy loops and `RecordConverter`) re-apply it on every `Struct.get()`. 2. **A null that reaches the sink is a value, not a gap.** Only the deserializing converter can tell a field that was absent on the wire from one explicitly set to null, and that resolution is finished by the time a `Struct` exists (an Avro record cannot even omit a field of its writer schema; the JSON converter resolves nulls according to its own `replace.null.with.default`). Past that boundary, the sink has no basis to treat a null as "missing" — substituting the DDL default fabricates a value the source row never contained. ### Reproduction Minimal demonstration of the root API behavior (connect-api 3.9.2): ```java Schema schema = SchemaBuilder.struct() .field("receiver_name", SchemaBuilder.string().optional().defaultValue("").build()) .build(); Struct struct = new Struct(schema); struct.put("receiver_name", null); struct.get("receiver_name"); // "" — default applied struct.getWithoutDefault("receiver_name"); // null ``` Any record carrying an explicit null in a field whose schema has a default reproduces the end-to-end corruption: pass it through `DebeziumTransform` (or directly into the sink) and the table receives the default instead of `null`. ### Precedent The same bug class has been fixed across the ecosystem, in two patterns: - **Debezium** uses `getWithoutDefault` unconditionally in its own SMTs: [`ExtractNewRecordState#L191-L193`](https://github.com/debezium/debezium/blob/v3.2.0.Final/debezium-core/src/main/java/io/debezium/transforms/ExtractNewRecordState.java#L191-L193) (with the comment *"Using get method may perform unwanted manipulation for the value (e.g: replacing null value with default value)"*) and [`ExtractChangedRecordState#L88-L89`](https://github.com/debezium/debezium/blob/v3.2.0.Final/debezium-core/src/main/java/io/debezium/transforms/ExtractChangedRecordState.java#L88-L89). `DebeziumTransform` in this repo rebuilds the record the same way (copying payload fields into a new flat value), and for exactly that operation Debezium's own SMT preserves nulls while this one does not. - **Kafka core** ([KIP-1040](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1040%3A+Improve+handling+of+nullable+values+in+InsertField%2C+ExtractField%2C+and+other+transformations)) added a `replace.null.with.default` option (default `true`) to nine generic SMTs (`InsertField`, `ExtractField`, `Cast`, …); [KIP-581](https://cwiki.apache.org/confluence/display/KAFKA/KIP-581%3A+Value+of+optional+null+field+which+has+default+value) added the same knob to `JsonConverter`. - **JDBC sinks**: [confluentinc/kafka-connect-jdbc#1433](https://github.com/confluentinc/kafka-connect-jdbc/pull/1433) added `replace.null.with.default` (default `true`); debezium-connector-jdbc fixed the same issue in [#50](https://github.com/debezium/debezium-connector-jdbc/pull/50) ([DBZ-7191](https://issues.redhat.com/browse/DBZ-7191)). ### Proposed fix 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 should therefore be pure pass-throughs, and the single behavioral decision should happen at the final read in `RecordConverter`, controlled by one sink option: 1. **SMTs (`DebeziumTransform`, `KafkaMetadataTransform`, `CopyValue`): switch to `getWithoutDefault` unconditionally.** This matches Debezium's own `ExtractNewRecordState`, and is not an observable behavior change on the default path: the output schema keeps the `defaultValue`, so any downstream consumer reading with `get()` (including today's `RecordConverter`) receives exactly the same values as before. Per-SMT options (the KIP-1040 pattern) were considered but not preferred: a single SMT in the chain left at `true` silently defeats the sink-level option, turning one bug into a configuration puzzle. That said, if maintainers prefer the KIP-1040 pattern (a `replace.null.with.default` option per SMT, default `true`), we are happy to implement that instead; the sink-level option below composes with either choice. 2. **`RecordConverter`: read struct fields via a shared helper gated by a new option `iceberg.tables.replace-null-with-default`, default `true`** (current behavior preserved; set to `false` to keep explicit nulls). The `true` default is proposed purely for backward compatibility — given point 2 above (a null that reaches the sink is a value, not a gap), maintainers may prefer `false`; happy to go either way. 3. **`RecordUtils.extractFromRecordValue` (route-field extraction): gate the same read with the same option.** The sink already skips records whose route value is null (`SinkWriter`), so with `replace-null-with-default=false` an explicitly-null route field is simply treated like any other null route value. Leaving this read ungated would make the option inconsistent: writes would preserve nulls while routing would still resolve defaults. 4. README config row + unit tests covering the transform, converter, and routing paths. The docs will note that (a) with the option disabled, an explicitly-null route field is skipped like any other null route value, and (b) pipelines carrying schemas via the JSON converter must also disable the converter's own `replace.null.with.default`, since that substitution happens at (de)serialization, outside the sink's reach (and on the producing side as well, if JSON is used there). ### Willingness to contribute - [x] I can contribute a fix for this bug independently - [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community - [ ] I cannot contribute a fix for this bug at this time -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
