[KAFKA-15838] [Connect] ExtractField and InsertField NULL Values are replaced by default value even in NULLABLE fields

• Type: Bug
• Status: In Progress
• Priority: Major
• Resolution: Unresolved
• Component/s: connect
• Assignee: Mario Fiore Vitale
• Reporter: Eric Pangiawan
• Created: 16/Nov/23 04:13
• Updated: 26/Apr/24 13:42

Description

ExtractField: Line 116-119

https://github.com/a0x8o/kafka/blob/master/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/ExtractField.java#L61-L68

InsertField: Line 163 - 195

https://github.com/a0x8o/kafka/blob/master/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertField.java#L163-L195

Expect:

Value `null` is valid for an optional filed, even though the field has a default value.
Only when field is required, the class return default value fallback when value is `null`.

Actual:

Always return default value if `null` was given.

Example:

PostgreSQL DDL:

CREATE TABLE products(
    id varchar(255),
    color varchar(255),
    quantity float8
);
-- Set Default
ALTER TABLE products ALTER COLUMN quantity SET  DEFAULT 1.0; 

Insert A Record:

INSERT INTO public.products VALUES('1', 'Blue', null); 

Table Select *:

 id | color | quantity
----+-------+----------
 1  | Blue  | 

Debezium Behavior when using ExtractField and InsertField class (in the event flattening SMT):

{
    "id":"1",
    "color":"Blue",
    "quantity":1.0,
    "__op":"c",
    "__ts_ms":1698127432079,
    "__source_ts_ms":1698127431679,
    "__db":"testing_db",
    "__schema":"public",
    "__table":"products",
    "__lsn":24470112,
    "__txId":957,
    "__snapshot":null,
    "__deleted":"false"
} 

The debezium code can be found here

Expected Output:

{
    "id":"1",
    "color":"Blue",
    "quantity":null,
    "__op":"c",
    "__ts_ms":1698127432079,
    "__source_ts_ms":1698127431679,
    "__db":"testing_db",
    "__schema":"public",
    "__table":"products",
    "__lsn":24470112,
    "__txId":957,
    "__snapshot":null,
    "__deleted":"false"
}

Temporary Solution:

use getWithoutDefault() into ExtractField and InsertField instead of get()

Issue Links

• Blocked: KAFKA-8713 [Connect] JsonConverter NULL Values are replaced by default values even in NULLABLE fields
• links to: GitHub Pull Request #15756