如何通过Flink Table API向Kafka Sink发送自定义Headers？

目前Flink SQL本身没有直接通过DDL配置自定义Headers的语法，但可以通过以下两种方式实现：

方式一：使用UDF结合Kafka动态表属性 #

• 定义生成Headers的UDF，返回Map<String, String>类型：

public class GenerateHeaders extends ScalarFunction {
    public Map<String, String> eval(String someColumn) {
        Map<String, String> headers = new HashMap<>();
        headers.put("user-id", someColumn);
        headers.put("event-time", String.valueOf(System.currentTimeMillis()));
        return headers;
    }
}

• 在Flink SQL中注册该UDF：

CREATE FUNCTION generate_headers AS 'com.yourpackage.GenerateHeaders';

• 创建Kafka Sink时，通过sink.kafka.headers.map属性指定存储Headers的字段：

CREATE TABLE kafka_sink (
    id STRING,
    content STRING,
    custom_headers MAP<STRING, STRING>
) WITH (
    'connector' = 'kafka',
    'topic' = 'your_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'sink.kafka.headers.map' = 'custom_headers'
);

• 写入数据时调用UDF生成Headers：

INSERT INTO kafka_sink
SELECT id, content, generate_headers(id) AS custom_headers
FROM your_source_table;

方式二：自定义Kafka序列化器 #

如果需要更灵活的Headers控制，可以自定义KafkaSerializationSchema实现：

• 实现自定义序列化器，在逻辑中添加Headers：

public class CustomKafkaSerializer implements KafkaSerializationSchema<RowData> {
    @Override
    public ProducerRecord<byte[], byte[]> serialize(RowData element, KafkaSinkContext context, Long timestamp) {
        // 自行实现value的序列化逻辑
        byte[] value = ...;
        // 添加自定义Headers
        Headers headers = new RecordHeaders();
        headers.add("custom-header-1", "value1".getBytes(StandardCharsets.UTF_8));
        headers.add("custom-header-2", element.getString(0).getBytes(StandardCharsets.UTF_8));
        return new ProducerRecord<>("your_topic", null, timestamp, null, value, headers);
    }
}

• 在Flink SQL的Sink配置中指定该序列化器（Flink 1.13+版本支持）：

CREATE TABLE kafka_sink (
    id STRING,
    content STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'your_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'table.exec.sink.kafka.serialization.schema' = 'com.yourpackage.CustomKafkaSerializer'
);

注意：第二种方式需要将自定义序列化器打包到作业的classpath中，确保Flink集群可以加载到该类。

内容的提问来源于stack exchange，提问作者Houston