[doc] Added metadata_column parameter usage example in mysql cdc action

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

@@ -51,7 +51,7 @@ To use this feature through `flink run`, run the following shell command.
     [--primary_keys <primary-keys>] \
     [--type_mapping <option1,option2...>] \
     [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
-    [--metadata_column <metadata-column>] \
+    [--metadata_columns <metadata-columns>] \
     [--mysql_conf <mysql-cdc-source-conf> [--mysql_conf <mysql-cdc-source-conf> ...]] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
     [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]
@@ -135,7 +135,7 @@ To use this feature through `flink run`, run the following shell command.
     [--including_tables <mysql-table-name|name-regular-expr>] \
     [--excluding_tables <mysql-table-name|name-regular-expr>] \
     [--mode <sync-mode>] \
-    [--metadata_column <metadata-column>] \
+    [--metadata_columns <metadata-columns>] \
     [--type_mapping <option1,option2...>] \
     [--mysql_conf <mysql-cdc-source-conf> [--mysql_conf <mysql-cdc-source-conf> ...]] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \

docs/layouts/shortcodes/generated/mysql_sync_database.html

@@ -62,8 +62,8 @@
         <td>It is used to specify synchronization mode.<br />Possible values:<ul><li>"divided" (the default mode if you haven't specified one): start a sink for each table, the synchronization of the new table requires restarting the job.</li><li>"combined": start a single combined sink for all tables, the new table will be automatically synchronized.</li></ul></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, such as the `table_name`, `database_name`, and `op_ts`. Each configuration should be specified in the format "key=value". See its <a href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#available-metadata">document</a> for a complete list of available metadata.</td>
+        <td><h5>--metadata_columns</h5></td>
+        <td>--metadata_columns 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, for example "table_name,database_name,op_ts". See its <a href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#available-metadata">document</a> for a complete list of available metadata.</td>
     </tr>
     <tr>
         <td><h5>--type_mapping</h5></td>

docs/layouts/shortcodes/generated/mysql_sync_table.html

@@ -66,8 +66,8 @@
         <td>The definitions of computed columns. The argument field is from MySQL 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, such as the `table_name`, `database_name`, and `op_ts`. Each configuration should be specified in the format "key=value". See its <a href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#available-metadata">document</a> for a complete list of available metadata.</td>
+        <td><h5>--metadata_columns</h5></td>
+        <td>--metadata_columns 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, for example "table_name,database_name,op_ts". See its <a href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#available-metadata">document</a> for a complete list of available metadata.</td>
     </tr>
     <tr>
         <td><h5>--mysql_conf</h5></td>

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

@@ -61,7 +61,7 @@ public class CdcActionCommonUtils {
     public static final String PARTITION_KEYS = "partition_keys";
     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_COLUMNS = "metadata_columns";
 
     public static void assertSchemaCompatible(
             TableSchema paimonSchema, List<DataField> sourceTableFields) {

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

@@ -32,6 +32,7 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -76,6 +77,10 @@ public SynchronizationActionBase withTypeMapping(TypeMapping typeMapping) {
         return this;
     }
 
+    public SynchronizationActionBase withMetadataColumns(String... metadataColumns) {
+        return withMetadataColumns(Arrays.asList(metadataColumns));
+    }
+
     public SynchronizationActionBase withMetadataColumns(List<String> metadataColumns) {
         this.metadataConverters =
                 metadataColumns.stream()

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

@@ -24,12 +24,11 @@
 import org.apache.paimon.flink.action.MultipleParameterToolAdapter;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
 
-import java.util.Arrays;
 import java.util.Optional;
 
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.EXCLUDING_TABLES;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.INCLUDING_TABLES;
-import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.METADATA_COLUMN;
+import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.METADATA_COLUMNS;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.MYSQL_CONF;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.TABLE_PREFIX;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.TABLE_SUFFIX;
@@ -69,8 +68,9 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                 .includingTables(params.get(INCLUDING_TABLES))
                 .excludingTables(params.get(EXCLUDING_TABLES))
                 .withMode(MultiTablesSinkMode.fromString(params.get(MODE)));
-        if (params.has(METADATA_COLUMN)) {
-            action.withMetadataColumns(Arrays.asList(params.get(METADATA_COLUMN).split(",")));
+
+        if (params.has(METADATA_COLUMNS)) {
+            action.withMetadataColumns(params.get(METADATA_COLUMNS).split(","));
         }
 
         if (params.has(TYPE_MAPPING)) {
@@ -101,7 +101,7 @@ public void printHelp() {
                         + "[--including_tables <mysql_table_name|name_regular_expr>] "
                         + "[--excluding_tables <mysql_table_name|name_regular_expr>] "
                         + "[--mode <sync_mode>] "
-                        + "[--metadata_column <metadata_column>] "
+                        + "[--metadata_columns <metadata_columns>] "
                         + "[--type_mapping <option1,option2...>] "
                         + "[--mysql_conf <mysql_cdc_source_conf> [--mysql_conf <mysql_cdc_source_conf> ...]] "
                         + "[--catalog_conf <paimon_catalog_conf> [--catalog_conf <paimon_catalog_conf> ...]] "
@@ -147,7 +147,7 @@ public void printHelp() {
         System.out.println();
 
         System.out.println(
-                "--metadata_column is used to specify which metadata columns to include in the output schema of the connector. Please see the doc for usage.");
+                "--metadata_columns is used to specify which metadata columns to include in the output schema of the connector. Please see the doc for usage.");
         System.out.println();
 
         System.out.println(

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

@@ -29,7 +29,7 @@
 import java.util.Optional;
 
 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_COLUMNS;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.MYSQL_CONF;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PARTITION_KEYS;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PRIMARY_KEYS;
@@ -73,8 +73,8 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                     new ArrayList<>(params.getMultiParameter(COMPUTED_COLUMN)));
         }
 
-        if (params.has(METADATA_COLUMN)) {
-            action.withMetadataColumns(new ArrayList<>(params.getMultiParameter(METADATA_COLUMN)));
+        if (params.has(METADATA_COLUMNS)) {
+            action.withMetadataColumns(params.get(METADATA_COLUMNS).split(","));
         }
 
         if (params.has(TYPE_MAPPING)) {
@@ -100,7 +100,7 @@ public void printHelp() {
                         + "[--primary_keys <primary_keys>] "
                         + "[--type_mapping <option1,option2...>] "
                         + "[--computed_column <'column_name=expr_name(args[, ...])'> [--computed_column ...]] "
-                        + "[--metadata_column <metadata_column>] "
+                        + "[--metadata_columns <metadata_columns>] "
                         + "[--mysql_conf <mysql_cdc_source_conf> [--mysql_conf <mysql_cdc_source_conf> ...]] "
                         + "[--catalog_conf <paimon_catalog_conf> [--catalog_conf <paimon_catalog_conf> ...]] "
                         + "[--table_conf <paimon_table_sink_conf> [--table_conf <paimon_table_sink_conf> ...]]");
@@ -126,7 +126,7 @@ public void printHelp() {
         System.out.println();
 
         System.out.println(
-                "--metadata_column is used to specify which metadata columns to include in the output schema of the connector. Please see the doc for usage.");
+                "--metadata_columns is used to specify which metadata columns to include in the output schema of the connector. Please see the doc for usage.");
         System.out.println();
 
         System.out.println("MySQL CDC source conf syntax:");