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 a6510ee05bbe62c3ac1a2f8f2be5427fec1317e1 Author: Arvid Heise <[email protected]> AuthorDate: Thu Jun 4 09:20:49 2026 +0200 [FLINK-39792] Add header-list metadata key to Kafka connector The existing headers metadata key (MAP<STRING, BYTES>) silently drops duplicate header keys and loses insertion order, since iteration into a HashMap lets the last value win. The new header-list key exposes headers as ARRAY<ROW<key STRING, value BYTES>>, preserving wire order and duplicate keys faithfully on both source and sink sides. The two keys are mutually exclusive on the write path: declaring both in the same table schema is rejected at planning time with a ValidationException. Null header values are preserved; null header keys are silently dropped. Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --- docs/content/docs/connectors/table/kafka.md | 10 +- .../DynamicKafkaRecordSerializationSchema.java | 15 ++- .../connectors/kafka/table/KafkaDynamicSink.java | 52 ++++++++ .../connectors/kafka/table/KafkaDynamicSource.java | 35 +++++ .../DynamicKafkaRecordSerializationSchemaTest.java | 3 +- .../kafka/table/KafkaDynamicTableFactoryTest.java | 47 +++++++ .../connectors/kafka/table/KafkaTableITCase.java | 144 +++++++++++++++++++++ 7 files changed, 302 insertions(+), 4 deletions(-) diff --git a/docs/content/docs/connectors/table/kafka.md b/docs/content/docs/connectors/table/kafka.md index fab543cd..d73235f4 100644 --- a/docs/content/docs/connectors/table/kafka.md +++ b/docs/content/docs/connectors/table/kafka.md @@ -93,8 +93,14 @@ Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT I </tr> <tr> <td><code>headers</code></td> - <td><code>MAP<STRING, BYTES> NOT NULL</code></td> - <td>Headers of the Kafka record as a map of raw bytes.</td> + <td><code>MAP<STRING, BYTES> NOT NULL</code></td> + <td>Headers of the Kafka record as a map. Duplicate keys are collapsed (last value wins) and insertion order is lost.</td> + <td><code>R/W</code></td> + </tr> + <tr> + <td><code>header-list</code></td> + <td><code>ARRAY<ROW<key STRING, value BYTES>> NOT NULL</code></td> + <td>Headers of the Kafka record as an ordered list of key-value pairs. Preserves duplicate keys and insertion order. Mutually exclusive with <code>headers</code> on the write path.</td> <td><code>R/W</code></td> </tr> <tr> diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java index febfa533..79381a38 100644 --- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java +++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java @@ -37,6 +37,7 @@ import org.apache.flink.types.RowKind; import org.apache.flink.util.Preconditions; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Header; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -171,7 +172,9 @@ class DynamicKafkaRecordSerializationSchema readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.TIMESTAMP), keySerialized, valueSerialized, - readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.HEADERS)); + resolveHeaders( + readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.HEADERS), + readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.HEADER_LIST))); } @Override @@ -285,4 +288,14 @@ class DynamicKafkaRecordSerializationSchema } return (T) metadata.converter.read(consumedRow, pos); } + + private static @Nullable List<Header> resolveHeaders( + @Nullable List<Header> mapHeaders, @Nullable List<Header> arrayHeaders) { + if (mapHeaders != null && arrayHeaders != null) { + throw new IllegalStateException( + "Both 'headers' and 'header-list' metadata keys are active simultaneously. " + + "This should have been caught during planning."); + } + return mapHeaders != null ? mapHeaders : arrayHeaders; + } } 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 4a1a4dd7..c15dac2c 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 @@ -31,6 +31,7 @@ import org.apache.flink.connector.kafka.sink.TransactionNamingStrategy; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSink; import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.Projection; import org.apache.flink.table.connector.ProviderContext; @@ -274,6 +275,13 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada @Override public void applyWritableMetadata(List<String> metadataKeys, DataType consumedDataType) { + if (metadataKeys.contains(WritableMetadata.HEADERS.key) + && metadataKeys.contains(WritableMetadata.HEADER_LIST.key)) { + throw new ValidationException( + String.format( + "The writable metadata is ambiguous. Please use either metadata key '%s' or '%s'.", + WritableMetadata.HEADERS.key, WritableMetadata.HEADER_LIST.key)); + } this.metadataKeys = metadataKeys; this.consumedDataType = consumedDataType; } @@ -448,6 +456,8 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada final List<Header> headers = new ArrayList<>(); for (int i = 0; i < keyArray.size(); i++) { if (!keyArray.isNullAt(i) && !valueArray.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); headers.add(new KafkaHeader(key, value)); @@ -457,6 +467,48 @@ public class KafkaDynamicSink implements DynamicTableSink, SupportsWritingMetada } }), + /** + * Wire-faithful alternative to {@code headers}: preserves duplicate keys and insertion + * order using {@code ARRAY<ROW<key STRING, value BYTES>>} instead of a lossy MAP. + */ + HEADER_LIST( + "header-list", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "key", DataTypes.STRING().nullable()), + DataTypes.FIELD( + "value", DataTypes.BYTES().nullable())) + .nullable()) + .nullable(), + new MetadataConverter() { + private static final long serialVersionUID = 1L; + + @Override + public Object read(RowData row, int pos) { + if (row.isNullAt(pos)) { + return null; + } + final ArrayData arrayData = row.getArray(pos); + final List<Header> headers = new ArrayList<>(); + for (int i = 0; i < arrayData.size(); i++) { + if (!arrayData.isNullAt(i)) { + final RowData headerRow = arrayData.getRow(i, 2); + if (headerRow.isNullAt(0)) { + continue; + } + // StringData.toString() decodes UTF-8; invalid bytes produce + // replacement characters (see FLIP-568). + final String name = headerRow.getString(0).toString(); + final byte[] value = + headerRow.isNullAt(1) ? null : headerRow.getBinary(1); + headers.add(new KafkaHeader(name, value)); + } + } + return headers; + } + }), + TIMESTAMP( "timestamp", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3).nullable(), diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java index 8c86320e..39f7b014 100644 --- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java +++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java @@ -44,7 +44,9 @@ import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata; import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown; +import org.apache.flink.table.data.GenericArrayData; import org.apache.flink.table.data.GenericMapData; +import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; @@ -623,6 +625,39 @@ public class KafkaDynamicSource } }), + /** + * Wire-faithful alternative to {@code headers}: preserves duplicate keys and insertion + * order. Key is NOT NULL; headers with a null key are silently omitted. + */ + HEADER_LIST( + "header-list", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "key", DataTypes.STRING().notNull()), + DataTypes.FIELD( + "value", DataTypes.BYTES().nullable())) + .notNull()) + .notNull(), + new MetadataConverter() { + private static final long serialVersionUID = 1L; + + @Override + public Object read(ConsumerRecord<?, ?> record) { + final List<GenericRowData> rows = new ArrayList<>(); + for (Header header : record.headers()) { + if (header.key() == null) { + continue; + } + final GenericRowData row = new GenericRowData(2); + row.setField(0, StringData.fromString(header.key())); + row.setField(1, header.value()); + rows.add(row); + } + return new GenericArrayData(rows.toArray()); + } + }), + LEADER_EPOCH( "leader-epoch", DataTypes.INT().nullable(), diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchemaTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchemaTest.java index 5a083927..de5e4c9d 100644 --- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchemaTest.java +++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchemaTest.java @@ -118,7 +118,8 @@ class DynamicKafkaRecordSerializationSchemaTest { public void open(InitializationContext context) throws Exception {} }; - int[] metadataPositions = new int[3]; + int[] metadataPositions = new int[KafkaDynamicSink.WritableMetadata.values().length]; + Arrays.fill(metadataPositions, -1); metadataPositions[KafkaDynamicSink.WritableMetadata.TOPIC.ordinal()] = 1; metadataPositions[KafkaDynamicSink.WritableMetadata.HEADERS.ordinal()] = 2; metadataPositions[KafkaDynamicSink.WritableMetadata.TIMESTAMP.ordinal()] = 3; diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java index 3c93112b..5f309344 100644 --- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java +++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java @@ -709,6 +709,7 @@ class KafkaDynamicTableFactoryTest { assertThat(actualKafkaSink.listWritableMetadata()) .containsOnlyKeys( KafkaDynamicSink.WritableMetadata.HEADERS.key, + KafkaDynamicSink.WritableMetadata.HEADER_LIST.key, KafkaDynamicSink.WritableMetadata.TIMESTAMP.key); } @@ -969,6 +970,7 @@ class KafkaDynamicTableFactoryTest { .containsOnlyKeys( KafkaDynamicSink.WritableMetadata.TOPIC.key, KafkaDynamicSink.WritableMetadata.HEADERS.key, + KafkaDynamicSink.WritableMetadata.HEADER_LIST.key, KafkaDynamicSink.WritableMetadata.TIMESTAMP.key); } @@ -1008,6 +1010,7 @@ class KafkaDynamicTableFactoryTest { .containsOnlyKeys( KafkaDynamicSink.WritableMetadata.TOPIC.key, KafkaDynamicSink.WritableMetadata.HEADERS.key, + KafkaDynamicSink.WritableMetadata.HEADER_LIST.key, KafkaDynamicSink.WritableMetadata.TIMESTAMP.key); } @@ -1323,6 +1326,50 @@ class KafkaDynamicTableFactoryTest { assertKafkaSource(provider); } + @Test + void testHeadersAndMultiHeadersMutuallyExclusive() { + final KafkaDynamicSink sink = + (KafkaDynamicSink) createTableSink(SCHEMA, getBasicSinkOptions()); + assertThatThrownBy( + () -> + sink.applyWritableMetadata( + Arrays.asList( + KafkaDynamicSink.WritableMetadata.HEADERS.key, + KafkaDynamicSink.WritableMetadata.HEADER_LIST.key), + SCHEMA.toSinkRowDataType())) + .isInstanceOf(ValidationException.class) + .hasMessageContaining(KafkaDynamicSink.WritableMetadata.HEADERS.key) + .hasMessageContaining(KafkaDynamicSink.WritableMetadata.HEADER_LIST.key); + } + + @Test + void testHeaderListMetadataTypes() { + assertThat(KafkaDynamicSource.ReadableMetadata.HEADER_LIST.dataType) + .isEqualTo( + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "key", + DataTypes.STRING().notNull()), + DataTypes.FIELD( + "value", + DataTypes.BYTES().nullable())) + .notNull()) + .notNull()); + assertThat(KafkaDynamicSink.WritableMetadata.HEADER_LIST.dataType) + .isEqualTo( + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "key", + DataTypes.STRING().nullable()), + DataTypes.FIELD( + "value", + DataTypes.BYTES().nullable())) + .nullable()) + .nullable()); + } + @Test void testDisableDiscoverPartition() { Map<String, String> tableSourceOptions = 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 e90c2191..ed0cafed 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 @@ -49,6 +49,7 @@ import org.assertj.core.api.ThrowingConsumer; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; import java.time.Instant; @@ -760,6 +761,149 @@ class KafkaTableITCase extends KafkaTableTestBase { cleanupTopic(topic); } + @ParameterizedTest(name = "format: {0}") + @MethodSource("formats") + void testKafkaSourceSinkWithMultiHeaders(final String format) throws Exception { + final String topic = "header_list_topic_" + format + "_" + UUID.randomUUID(); + createTestTopic(topic, 1, 1); + + String groupId = getStandardProps().getProperty("group.id"); + String bootstraps = getBootstrapServers(); + + final String createTable = + String.format( + "CREATE TABLE kafka (\n" + + " `physical_1` STRING,\n" + + " `header-list` ARRAY<ROW<`key` STRING, `value` 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" + + " %s\n" + + ")", + topic, bootstraps, groupId, formatOptions(format)); + tEnv.executeSql(createTable); + + // Row 1: duplicate key 'k1' — both entries must survive the round-trip. + // Row 2: null header array. + // Row 3: null-key entry is silently dropped; null value on k3 is preserved. + // Row 4: null ROW in array is silently skipped; adjacent entries survive. + tEnv.executeSql( + "INSERT INTO kafka VALUES\n" + + " ('data 1', ARRAY[ROW('k1', X'C0FFEE'), ROW('k1', X'BABE01')], TRUE),\n" + + " ('data 2', CAST(NULL AS ARRAY<ROW<`key` STRING, `value` BYTES>>), FALSE),\n" + + " ('data 3', ARRAY[ROW('k1', X'102030'), ROW(CAST(NULL AS STRING), X'DEAD')," + + " ROW('k2', X'203040'), ROW('k3', CAST(NULL AS BYTES))], TRUE),\n" + + " ('data 4', ARRAY[ROW('ka', X'0102')," + + " CAST(NULL AS ROW<`key` STRING, `value` BYTES>)," + + " ROW('kb', CAST(NULL AS BYTES))], FALSE)") + .await(); + + final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 4); + + final List<Row> expected = + Arrays.asList( + Row.of( + "data 1", + new Row[] { + Row.of("k1", EncodingUtils.decodeHex("C0FFEE")), + Row.of("k1", EncodingUtils.decodeHex("BABE01")) + }, + true), + // null and empty arrays are indistinguishable on the wire; source returns + // []. + Row.of("data 2", new Row[0], false), + Row.of( + "data 3", + new Row[] { + Row.of("k1", EncodingUtils.decodeHex("102030")), + Row.of("k2", EncodingUtils.decodeHex("203040")), + Row.of("k3", null) + }, + true), + Row.of( + "data 4", + new Row[] { + Row.of("ka", EncodingUtils.decodeHex("0102")), + Row.of("kb", null) + }, + false)); + + assertThat(result).satisfies(matching(deepEqualTo(expected, true))); + + cleanupTopic(topic); + } + + /** + * Verifies that {@code headers} and {@code header-list} are interoperable: data written via one + * can be read back via the other, and both columns can be declared on the same source. + */ + @ParameterizedTest(name = "writeMultiHeaders: {0}") + @ValueSource(booleans = {true, false}) + void testKafkaHeadersInteroperability(final boolean writeMulti) throws Exception { + final String topic = "headers_interop_topic_" + writeMulti + "_" + UUID.randomUUID(); + createTestTopic(topic, 1, 1); + + final String groupId = getStandardProps().getProperty("group.id"); + final String bootstraps = getBootstrapServers(); + + final String hVirtual = writeMulti ? " VIRTUAL" : ""; + final String mhVirtual = writeMulti ? "" : " VIRTUAL"; + + tEnv.executeSql( + String.format( + "CREATE TABLE kafka (\n" + + " `physical_1` STRING,\n" + + " `headers` MAP<STRING, BYTES> METADATA%s,\n" + + " `header-list` ARRAY<ROW<`key` STRING, `value` BYTES>> METADATA%s,\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" + + ")", + hVirtual, mhVirtual, topic, bootstraps, groupId)); + + final String dataValues = + writeMulti ? "ARRAY[ROW('k1', X'C0FFEE')]" : "MAP['k1', X'C0FFEE']"; + final String nullValues = + writeMulti + ? "CAST(NULL AS ARRAY<ROW<`key` STRING, `value` BYTES>>)" + : "CAST(NULL AS MAP<STRING, BYTES>)"; + + tEnv.executeSql( + "INSERT INTO kafka VALUES\n" + + " ('data 1', " + + dataValues + + ", TRUE),\n" + + " ('data 2', " + + nullValues + + ", FALSE)") + .await(); + + final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 2); + + final List<Row> expected = + Arrays.asList( + Row.of( + "data 1", + map(entry("k1", EncodingUtils.decodeHex("C0FFEE"))), + new Row[] {Row.of("k1", EncodingUtils.decodeHex("C0FFEE"))}, + true), + // null/empty → no headers on wire; both columns read back as empty. + Row.of("data 2", Collections.emptyMap(), new Row[0], false)); + + assertThat(result).satisfies(matching(deepEqualTo(expected, true))); + + cleanupTopic(topic); + } + @ParameterizedTest(name = "format: {0}") @MethodSource("formats") void testKafkaSourceSinkWithKeyAndPartialValue(final String format) throws Exception {
