[cdc] Add metadata column support for Kafka CDC connector (#7315)

This PR supercedes #6353 - I took
over this work from my colleague @gmdfalk

## Description
Add --metadata_column support to Paimon Kafka CDC connector, similar to
the already existing options added for MySQL and Postgres:
#2077
Also add optional --metadata_column_prefix to avoid conflicts with
existing Paimon fields like topic, timestamp etc.

Supported metadata columns are those on
org.apache.kafka.clients.consumer.ConsumerRecord i.e.:

- topic
- partition
- offset
- timestamp
- timestampType: This is the name of the enum i.e. NoTimestampType,
CreateTime or LogAppendTime

The feature is backwards compatible. It's only active when
--metadata_column is supplied resp.
SynchronizationActionBase.withMetadataColumns is used.

## Motivation
This is a requested feature:
#3210

We primarly use this feature for two purposes:

1. Troubleshooting and data lineage (e.g. where in our Kafka
infrastructure does this Paimon row come from?)
2. Mapping large Kafka topic partitions 1:1 to Paimon buckets to avoid
reshuffling (see this issue it would solve:
#3249)

## Tests
Unit and Integration Tests

## API and Format
No changes to public apis or storage format.

The changes here are contained to the flink cdc package but I did have
to update CdcSourceRecord since it previously didn't provide a way to
surface arbitrary metadata for a record.

The metadata attribute on CdcSourceRecord is intentionally a generic Map
so that it can potentially be used to add metadata support for other
connectors like Pulsar or Mongo that are not yet implemented.

## Documentation
Added the new --metadata_column and --metadata_column_prefix parameter
to Kafka CDC docs.

## Dev notes
For running integration tests on MacOS with Rancher Desktop, i had to
properly expose the docker socket to testcontainers e.g. system wide via
sudo ln -sf "$HOME/.rd/docker.sock" /var/run/docker.sock.

---------

Signed-off-by: Max Falk <gfalk@yelp.com>
Co-authored-by: Max Falk <gfalk@yelp.com>
Co-authored-by: Max Falk <279131+gmdfalk@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

docs/content/cdc-ingestion/kafka-cdc.md

@@ -103,6 +103,8 @@ To use this feature through `flink run`, run the following shell command.
     [--primary_keys <primary-keys>] \
     [--type_mapping to-string] \
     [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
+    [--metadata_column <metadata-column> [--metadata_column ...]] \
+    [--metadata_column_prefix <metadata-column-prefix>] \
     [--kafka_conf <kafka-source-conf> [--kafka_conf <kafka-source-conf> ...]] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
     [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]
@@ -215,6 +217,8 @@ To use this feature through `flink run`, run the following shell command.
     [--partition_keys <partition_keys>] \
     [--primary_keys <primary-keys>] \
     [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
+    [--metadata_column <metadata-column> [--metadata_column ...]] \
+    [--metadata_column_prefix <metadata-column-prefix>] \
     [--kafka_conf <kafka-source-conf> [--kafka_conf <kafka-source-conf> ...]] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
     [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]

docs/layouts/shortcodes/generated/kafka_sync_database.html

@@ -94,6 +94,14 @@
         <td><h5>--computed_column</h5></td>
         <td>The definitions of computed columns. The argument field is from Kafka topic's table field name. See <a href="../overview/#computed-functions">here</a> for a complete list of configurations. NOTICE: It returns null if the referenced column does not exist in the source table.</td>
     </tr>
+    <tr>
+        <td><h5>--metadata_column</h5></td>
+        <td>--metadata_column is used to specify which metadata columns to include in the output schema of the connector. Metadata columns provide additional information related to the source data. Available values are topic, partition, offset, timestamp and timestamp_type (one of 'NoTimestampType', 'CreateTime' or 'LogAppendTime').</td>
+    </tr>
+    <tr>
+        <td><h5>--metadata_column_prefix</h5></td>
+        <td>--metadata_column_prefix is optionally used to set a prefix for metadata columns in the Paimon table to avoid conflicts with existing attributes. For example, with prefix "__kafka_", the metadata column "topic" will be stored as "__kafka_topic" field.</td>
+    </tr>
     <tr>
         <td><h5>--eager_init</h5></td>
         <td>It is default false. If true, all relevant tables commiter will be initialized eagerly, which means those tables could be forced to create snapshot.</td>

docs/layouts/shortcodes/generated/kafka_sync_table.html

@@ -70,6 +70,14 @@
         <td><h5>--computed_column</h5></td>
         <td>The definitions of computed columns. The argument field is from Kafka topic's table field name. See <a href="../overview/#computed-functions">here</a> for a complete list of configurations. </td>
     </tr>
+    <tr>
+        <td><h5>--metadata_column</h5></td>
+        <td>--metadata_column is used to specify which metadata columns to include in the output schema of the connector. Metadata columns provide additional information related to the source data. Available values are topic, partition, offset, timestamp and timestamp_type (one of 'NoTimestampType', 'CreateTime' or 'LogAppendTime').</td>
+    </tr>
+    <tr>
+        <td><h5>--metadata_column_prefix</h5></td>
+        <td>--metadata_column_prefix is optionally used to set a prefix for metadata columns in the Paimon table to avoid conflicts with existing attributes. For example, with prefix "__kafka_", the metadata column "topic" will be stored as "__kafka_topic" field.</td>
+    </tr>
     <tr>
         <td><h5>--kafka_conf</h5></td>
         <td>The configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic/topic-pattern`, `properties.group.id`,  and `value.format` are required configurations, others are optional.See its <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kafka/#connector-options">document</a> for a complete list of configurations.</td>
@@ -83,4 +91,4 @@
         <td>The configuration for Paimon table sink. Each configuration should be specified in the format "key=value". See <a href="{{ $ref }}">here</a> for a complete list of table configurations.</td>
     </tr>
     </tbody>
-</table>
+</table>

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java

@@ -68,6 +68,7 @@ public class CdcActionCommonUtils {
     public static final String PRIMARY_KEYS = "primary_keys";
     public static final String COMPUTED_COLUMN = "computed_column";
     public static final String METADATA_COLUMN = "metadata_column";
+    public static final String METADATA_COLUMN_PREFIX = "metadata_column_prefix";
     public static final String MULTIPLE_TABLE_PARTITION_KEYS = "multiple_table_partition_keys";
     public static final String EAGER_INIT = "eager_init";
     public static final String SYNC_PKEYS_FROM_SOURCE_SCHEMA =

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcMetadataConverter.java

@@ -34,13 +34,25 @@
  * A functional interface for converting CDC metadata.
  *
  * <p>This interface provides a mechanism to convert Change Data Capture (CDC) metadata from a given
- * {@link JsonNode} source. Implementations of this interface can be used to process and transform
- * metadata entries from CDC sources.
+ * {@link JsonNode} source or {@link CdcSourceRecord}. Implementations of this interface can be used
+ * to process and transform metadata entries from CDC sources.
  */
 public interface CdcMetadataConverter extends Serializable {
 
     String read(JsonNode payload);
 
+    /**
+     * Read metadata from a CDC source record. Default implementation throws
+     * UnsupportedOperationException to maintain backward compatibility.
+     *
+     * @param record the CDC source record
+     * @return the metadata value as a string
+     */
+    default String read(CdcSourceRecord record) {
+        throw new UnsupportedOperationException(
+                "This metadata converter does not support reading from CdcSourceRecord");
+    }
+
     DataType dataType();
 
     String columnName();

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcMetadataProcessor.java

@@ -18,6 +18,8 @@
 
 package org.apache.paimon.flink.action.cdc;
 
+import org.apache.paimon.flink.action.cdc.kafka.KafkaMetadataConverter;
+
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
 
 import java.util.Arrays;
@@ -49,7 +51,14 @@ public enum CdcMetadataProcessor {
             new CdcMetadataConverter.DatabaseNameConverter(),
             new CdcMetadataConverter.TableNameConverter(),
             new CdcMetadataConverter.SchemaNameConverter(),
-            new CdcMetadataConverter.OpTsConverter());
+            new CdcMetadataConverter.OpTsConverter()),
+    KAFKA_METADATA_PROCESSOR(
+            SyncJobHandler.SourceType.KAFKA,
+            new KafkaMetadataConverter.TopicConverter(),
+            new KafkaMetadataConverter.PartitionConverter(),
+            new KafkaMetadataConverter.OffsetConverter(),
+            new KafkaMetadataConverter.TimestampConverter(),
+            new KafkaMetadataConverter.TimestampTypeConverter());
 
     private final SyncJobHandler.SourceType sourceType;
 

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcSourceRecord.java

@@ -21,6 +21,9 @@
 import javax.annotation.Nullable;
 
 import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 
 /** A data change record from the CDC source. */
@@ -35,14 +38,29 @@ public class CdcSourceRecord implements Serializable {
     // TODO Use generics to support more scenarios.
     private final Object value;
 
+    // Generic metadata map - any source can add metadata
+    private final Map<String, Object> metadata;
+
     public CdcSourceRecord(@Nullable String topic, @Nullable Object key, Object value) {
-        this.topic = topic;
-        this.key = key;
-        this.value = value;
+        this(topic, key, value, null);
     }
 
     public CdcSourceRecord(Object value) {
-        this(null, null, value);
+        this(null, null, value, null);
+    }
+
+    public CdcSourceRecord(
+            @Nullable String topic,
+            @Nullable Object key,
+            Object value,
+            @Nullable Map<String, Object> metadata) {
+        this.topic = topic;
+        this.key = key;
+        this.value = value;
+        this.metadata =
+                metadata != null
+                        ? Collections.unmodifiableMap(new HashMap<>(metadata))
+                        : Collections.emptyMap();
     }
 
     @Nullable
@@ -59,6 +77,15 @@ public Object getValue() {
         return value;
     }
 
+    public Map<String, Object> getMetadata() {
+        return metadata;
+    }
+
+    @Nullable
+    public Object getMetadata(String key) {
+        return metadata.get(key);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (!(o instanceof CdcSourceRecord)) {
@@ -68,12 +95,13 @@ public boolean equals(Object o) {
         CdcSourceRecord that = (CdcSourceRecord) o;
         return Objects.equals(topic, that.topic)
                 && Objects.equals(key, that.key)
-                && Objects.equals(value, that.value);
+                && Objects.equals(value, that.value)
+                && Objects.equals(metadata, that.metadata);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(topic, key, value);
+        return Objects.hash(topic, key, value, metadata);
     }
 
     @Override

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/PrefixedMetadataConverter.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc;
+
+import org.apache.paimon.types.DataType;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * Wraps a {@link CdcMetadataConverter} to add a prefix to its column name.
+ *
+ * <p>This decorator allows adding prefixes like "__kafka_" to metadata column names to avoid
+ * collisions with source data columns, while keeping the underlying converter logic unchanged.
+ */
+public class PrefixedMetadataConverter implements CdcMetadataConverter {
+
+    private static final long serialVersionUID = 1L;
+
+    private final CdcMetadataConverter delegate;
+    private final String prefix;
+
+    public PrefixedMetadataConverter(CdcMetadataConverter delegate, String prefix) {
+        this.delegate = delegate;
+        this.prefix = prefix != null ? prefix : "";
+    }
+
+    @Override
+    public String columnName() {
+        return prefix + delegate.columnName();
+    }
+
+    @Override
+    public String read(JsonNode payload) {
+        return delegate.read(payload);
+    }
+
+    @Override
+    public String read(CdcSourceRecord record) {
+        return delegate.read(record);
+    }
+
+    @Override
+    public DataType dataType() {
+        return delegate.dataType();
+    }
+}

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java

@@ -207,9 +207,10 @@ public FlatMapFunction<CdcSourceRecord, RichCdcMultiplexRecord> provideRecordPar
                 return new PostgresRecordParser(
                         cdcSourceConfig, computedColumns, typeMapping, metadataConverters);
             case KAFKA:
+                return provideDataFormat()
+                        .createParser(typeMapping, computedColumns, metadataConverters);
             case PULSAR:
-                DataFormat dataFormat = provideDataFormat();
-                return dataFormat.createParser(typeMapping, computedColumns);
+                return provideDataFormat().createParser(typeMapping, computedColumns);
             case MONGODB:
                 return new MongoDBRecordParser(computedColumns, cdcSourceConfig);
             default:

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionFactoryBase.java

@@ -29,6 +29,7 @@
 
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.COMPUTED_COLUMN;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.METADATA_COLUMN;
+import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.METADATA_COLUMN_PREFIX;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PARTITION_KEYS;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PRIMARY_KEYS;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.SYNC_PKEYS_FROM_SOURCE_SCHEMA;
@@ -64,6 +65,11 @@ protected void withParams(MultipleParameterToolAdapter params, SyncTableActionBa
         }
 
         if (params.has(METADATA_COLUMN)) {
+            // Parse optional prefix first
+            if (params.has(METADATA_COLUMN_PREFIX)) {
+                action.withMetadataColumnPrefix(params.get(METADATA_COLUMN_PREFIX));
+            }
+
             List<String> metadataColumns =
                     new ArrayList<>(params.getMultiParameter(METADATA_COLUMN));
             if (metadataColumns.size() == 1) {