li3zhi4 opened a new issue, #11632: URL: https://github.com/apache/seatunnel/issues/11632
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found related but **different-root-cause** reports: - [#8759](https://github.com/apache/seatunnel/issues/8759) — Oracle CDC → Kafka: Debezium struct default value used when column is null (Debezium deserialization path). - [PR #7950](https://github.com/apache/seatunnel/commit/3b432125a) — fixed the Debezium path via `struct.getWithoutDefault(...)`; the **JSON format** deserializer (`seatunnel-format-json`) is a separate code path and was NOT touched. - [PR #6247](https://github.com/apache/seatunnel/pull/6247) — fixed `Object.class` option value parsing so `defaultValue` config can be read correctly. - No existing issue reports the JSON format deserializer ignoring `schema.columns[].defaultValue`. ### What happened When a source connector using the **JSON format** (Kafka, Pulsar, HTTP, MongoDB, Elasticsearch, …) configures `schema { columns = [ { name = "GROUPID", type = "int", nullable = false, defaultValue = 0 } ] }`, the `defaultValue` is **parsed and stored on the `Column`** but **never applied** during deserialization: if the JSON message is missing the field or contains an explicit `null`, the row field stays `null` instead of the configured default value. Expected: `defaultValue = 0` is used when the JSON field is missing or `null`. Actual: the field is `null`; the `defaultValue` config is silently ignored. ### SeaTunnel Version 2.3.13 (verified locally). **Also reproducible on current upstream `dev`** (checked 2026-08-03, `6bf786bee7`). ### SeaTunnel Config ```hocon source { Kafka { bootstrap.servers = "localhost:9092" topic = "test" format = json schema { columns = [ { name = "GROUPID", type = "int", nullable = false, defaultValue = 0 } { name = "name", type = "string", nullable = false, defaultValue = "unknown" } ] } } } sink { Console {} } ``` ### Running Command ```shell ./bin/seatunnel.sh --config ./config/xxx.conf ``` ### Error Exception No error. The job runs, but fields configured with `defaultValue` come out as `null` when the JSON message omits them or sets them to `null` — a silent data-integrity bug. ### Root cause (confirmed by code inspection on upstream dev) The `defaultValue` is lost on the way from `CatalogTable` to the row converter: ``` CatalogTable (contains Column metadata incl. defaultValue) ↓ getSeaTunnelRowType() SeaTunnelRowType (fieldNames[] + fieldTypes[] only) ← defaultValue dropped here ↓ JsonToRowConverters (no access to Column[] / defaultValue) ↓ convertField(): field == null → returns null (defaultValue never consulted) ``` Concretely, on upstream `dev` (same as 2.3.13): 1. `JsonToRowConverters.convertField(...)` (lines ~474-486) — for a missing field it only checks `failOnMissingField` and otherwise returns `null`; it has **no `Column[]` metadata** and never checks `getDefaultValue()`: ```java private Object convertField( JsonToObjectConverter fieldConverter, String fieldName, JsonNode field) { if (field == null) { if (failOnMissingField) { throw new IllegalArgumentException(...); } else { return null; // defaultValue never applied } } else { return fieldConverter.convert(field, fieldName); } } ``` 2. `JsonDeserializationSchema` (both constructors, lines ~82 and ~103) builds `new JsonToRowConverters(failOnMissingField, ignoreParseErrors)` and never passes `catalogTable.getTableSchema().getColumns()` to the converters. 3. The configuration side works correctly: `ReadonlyConfigParser` (line ~139) reads `columnConfig.get(ConnectorCommonOptions.DEFAULT_VALUE)` into `PhysicalColumn.of(...)`, and `Column.getDefaultValue()` exists in `seatunnel-api`. The API contract explicitly supports default values — the deserialization side just never uses them. ### Proposed fix (implemented & verified locally on 2.3.13) Minimal, backward-compatible, root-cause fix: - `JsonToRowConverters`: add a `Column[] columns` field and a new constructor `JsonToRowConverters(boolean failOnMissingField, boolean ignoreParseErrors, Column[] columns)`; add a `convertField(converter, fieldName, field, fieldIndex)` overload that consults `columns[fieldIndex].getDefaultValue()` when the field is missing or `null`, falling back to the original `failOnMissingField`/`null` behavior when no default value is configured. - `JsonDeserializationSchema`: extract `Column[]` from `catalogTable.getTableSchema().getColumns()` (with null-safety) and pass it to the new constructor. - New test `JsonDefaultValueTest`: missing field → defaultValue; explicit `null` → defaultValue; no defaultValue → `null` (backward compatibility); string-typed default values. Verification on 2.3.13: 3 new unit tests + full `seatunnel-format-json` module (46 tests) all pass; `spotless:apply` clean. ### Zeta or Flink or Spark Version zeta (format module is engine-agnostic; affects all of them) ### Java or Scala Version Java 8 / 11 (both OK) ### Screenshots _No response_ ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! (fix already implemented and verified on the custom 2.3.13 branch; will port to `dev`) ### Code of Conduct -- 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]
