This is an automated email from the ASF dual-hosted git repository. AHeise pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 25974e4d3900aff52af0ebc8a26712144791cabb Author: Arvid Heise <[email protected]> AuthorDate: Thu Jun 4 09:21:28 2026 +0200 [FLINK-39792] Fix headers MAP dropping null values The HEADERS converter skipped entries whose value was null (null bytes are legal in the Kafka wire protocol). The fix preserves null values as null byte arrays, which Kafka's producer accepts without error. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --- .../connectors/kafka/table/KafkaDynamicSink.java | 5 ++- .../connectors/kafka/table/KafkaTableITCase.java | 46 ++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java index c15dac2c..9c6c0fb2 100644 --- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java +++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java @@ -455,11 +455,12 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada final ArrayData valueArray = map.valueArray(); final List<Header> headers = new ArrayList<>(); for (int i = 0; i < keyArray.size(); i++) { - if (!keyArray.isNullAt(i) && !valueArray.isNullAt(i)) { + if (!keyArray.isNullAt(i)) { // StringData.toString() decodes UTF-8; invalid bytes produce // replacement characters (see FLIP-568). final String key = keyArray.getString(i).toString(); - final byte[] value = valueArray.getBinary(i); + final byte[] value = + valueArray.isNullAt(i) ? null : valueArray.getBinary(i); headers.add(new KafkaHeader(key, value)); } } diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java index ed0cafed..cff654f8 100644 --- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java +++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java @@ -47,6 +47,7 @@ import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import org.assertj.core.api.Assertions; import org.assertj.core.api.ThrowingConsumer; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; @@ -761,6 +762,51 @@ class KafkaTableITCase extends KafkaTableTestBase { cleanupTopic(topic); } + @Test + void testKafkaHeadersMapNullValue() throws Exception { + final String topic = "headers_null_value_topic_" + UUID.randomUUID(); + createTestTopic(topic, 1, 1); + + final String groupId = getStandardProps().getProperty("group.id"); + final String bootstraps = getBootstrapServers(); + + tEnv.executeSql( + String.format( + "CREATE TABLE kafka (\n" + + " `physical_1` STRING,\n" + + " `headers` MAP<STRING, BYTES> METADATA,\n" + + " `physical_2` BOOLEAN\n" + + ") WITH (\n" + + " 'connector' = 'kafka',\n" + + " 'topic' = '%s',\n" + + " 'properties.bootstrap.servers' = '%s',\n" + + " 'properties.group.id' = '%s',\n" + + " 'scan.startup.mode' = 'earliest-offset',\n" + + " 'format' = 'json'\n" + + ")", + topic, bootstraps, groupId)); + + tEnv.executeSql( + "INSERT INTO kafka VALUES ('data 1'," + + " MAP['k1', X'C0FFEE', 'k2', CAST(NULL AS BYTES)], TRUE)") + .await(); + + final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 1); + + final List<Row> expected = + Collections.singletonList( + Row.of( + "data 1", + map( + entry("k1", EncodingUtils.decodeHex("C0FFEE")), + entry("k2", null)), + true)); + + assertThat(result).satisfies(matching(deepEqualTo(expected, true))); + + cleanupTopic(topic); + } + @ParameterizedTest(name = "format: {0}") @MethodSource("formats") void testKafkaSourceSinkWithMultiHeaders(final String format) throws Exception {
