[
https://issues.apache.org/jira/browse/FLINK-39792?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser updated FLINK-39792:
-----------------------------------
Release Note: Adds the header-list metadata key, ARRAY<ROW<key STRING,
value BYTES>>, which keeps duplicate header keys and their order. It cannot be
combined with the headers key in one table. The headers key now keeps headers
whose value is null instead of dropping them.
> Add multi-headers metadata key to Kafka connector
> -------------------------------------------------
>
> Key: FLINK-39792
> URL: https://issues.apache.org/jira/browse/FLINK-39792
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Reporter: Arvid Heise
> Assignee: Arvid Heise
> Priority: Major
> Labels: pull-request-available
> Fix For: kafka-5.1.0
>
>
> Problem:
> The Kafka wire format models headers as an ordered, duplicate-key-allowed
> iterable ({{{}Headers extends Iterable<Header>{}}}), but the existing headers
> metadata key uses {{{}MAP<STRING, BYTES>{}}}, which cannot faithfully
> represent this: when multiple headers share a key, iteration into a
> {{HashMap}} silently discards all but the last value, losing both duplicates
> and insertion order. This makes the current metadata key unsuitable for use
> cases that rely on repeated header keys or order-sensitive header processing.
> Solution:
> Add a new metadata key {{multi-headers}} with type {{ARRAY<ROW<name STRING,
> value BYTES>>}} to {{KafkaDynamicSource}} and {{{}KafkaDynamicSink{}}}. On
> the source side, {{record.headers()}} is iterated in wire order into the
> array, preserving duplicates and position.
> On the sink side, the array is iterated and each entry is appended via
> {{{}record.headers().add(name, value){}}}. The existing ??headers?? key
> ({{{}MAP<STRING, BYTES>{}}}) is kept unchanged and soft-deprecated — no
> breaking change; users opt in to the new key at their own pace.
> If both, new and old header keys are used, we throw an error.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
