[jira] [Commented] (FLINK-39792) Add multi-headers metadata key to Kafka connector

["david radley (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-39792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18084039#comment-18084039 ] 

david radley commented on FLINK-39792:
--------------------------------------

[~arvid] this sounds like a good idea. I was looking around and it looks like the header value is always a string. Can we deprecate the amend the existing map and add 2 new ones using STRING instead of BYTE? This seems more accurate. Or is there a reason why we need BYTE here? 

> 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
>
> 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
(v8.20.10#820010)