add pipline support for kafka sink connector ,issue:datavane/tis#484

Author: baisui1981

.claude/settings.local.json

@@ -1,7 +1,9 @@
 {
   "permissions": {
     "allow": [
-      "Bash(find /opt/misc/flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-kingbase-cdc -name *.java -type f)"
+      "Bash(find /opt/misc/flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-kingbase-cdc -name *.java -type f)",
+      "Bash(find /Users/mozhenghua/j2ee_solution/project/plugins/tis-datax/tis-datax-kafka-plugin -name *.java -type f)",
+      "Bash(git -C /Users/mozhenghua/j2ee_solution/project/plugins diff HEAD -- tis-datax/tis-datax-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/datax/kafka/writer/DataXKafkaWriter.java)"
     ]
   }
 }

tis-datax/tis-datax-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/datax/kafka/writer/DataXKafkaWriter.java

@@ -19,7 +19,9 @@
 package com.qlangtech.tis.plugins.datax.kafka.writer;
 
 import com.alibaba.citrus.turbine.Context;
+import com.alibaba.datax.core.job.ISourceTable;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
 import com.qlangtech.tis.TIS;
 import com.qlangtech.tis.datax.IDataxContext;
 import com.qlangtech.tis.datax.IDataxProcessor;
@@ -33,8 +35,12 @@
 import com.qlangtech.tis.plugin.annotation.FormFieldType;
 import com.qlangtech.tis.plugin.annotation.Validator;
 import com.qlangtech.tis.plugin.datax.SelectedTab;
+import com.qlangtech.tis.plugin.datax.common.AutoCreateTable;
+import com.qlangtech.tis.plugin.datax.common.impl.NoneCreateTable;
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
+import com.qlangtech.tis.plugin.ds.IInitWriterTableExecutor;
 import com.qlangtech.tis.plugins.datax.kafka.writer.protocol.KafkaProtocol;
+import com.qlangtech.tis.plugins.incr.flink.cdc.pipeline.ICDCPipelineTableOptionsCreator;
 import com.qlangtech.tis.realtime.transfer.DTO;
 import com.qlangtech.tis.realtime.utils.NetUtils;
 import com.qlangtech.tis.runtime.module.misc.IControlMsgHandler;
@@ -51,17 +57,19 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.UnknownHostException;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
  * https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/kafka/ <br/>
  * https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json <br/>
  */
-public class DataXKafkaWriter extends DataxWriter {
+public class DataXKafkaWriter extends DataxWriter implements IInitWriterTableExecutor, ICDCPipelineTableOptionsCreator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DataXKafkaWriter.class);
 
@@ -138,6 +146,7 @@ public class DataXKafkaWriter extends DataxWriter {
     @FormField(ordinal = 22, type = FormFieldType.INT_NUMBER, validate = {Validator.require, Validator.integer}, advance = true)
     public Integer receiveBufferBytes;
 
+
     @Override
     public void startScanDependency() {
 
@@ -186,12 +195,35 @@ public Map<String, Object> buildKafkaConfig(boolean isTest) {
 //                .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName())//
                 .put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName())//   .StringSerializer.class.getName())
                 .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName())// JsonSerializer.class.getName())
+//                .put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArraySerializer")//   .StringSerializer.class.getName())
+//                .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArraySerializer")//
                 .build();
         return props.entrySet().stream()
                 .filter(entry -> entry.getValue() != null && StringUtils.isNotBlank(entry.getValue().toString()))
-                .collect(Collectors.toMap((e) -> e.getKey(), (e) -> String.valueOf(e.getValue())));
+                .collect(Collectors.toMap(Map.Entry::getKey, (e) -> String.valueOf(e.getValue())));
+    }
+
+//    @Override
+//    public EntityName parseEntity(ISelectedTab tab) {
+//        return EntityName.create(IEndTypeGetter.EndType.Kafka.getVal(), tab.getName());
+//    }
+
+    @Override
+    public AutoCreateTable getAutoCreateTableCanNotBeNull() {
+        return new NoneCreateTable();
     }
 
+    @Override
+    public void initWriterTable(ISourceTable sourceTable, String sinkTargetTabName, List<String> jdbcUrls) throws Exception {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Function<SelectedTab, Map<String, String>> createTabOpts() {
+        return (tab) -> {
+            return Maps.newHashMap();
+        };
+    }
 
     @TISExtension
     public static class DefaultDescriptor extends BaseDataxWriterDescriptor implements DataxWriter.IRewriteSuFormProperties {

tis-datax/tis-datax-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/datax/kafka/writer/KafkaSelectedTab.java

@@ -30,6 +30,7 @@
 import com.qlangtech.tis.util.IPluginContext;
 import com.qlangtech.tis.util.impl.AttrVals;
 
+import java.util.Collection;
 import java.util.List;
 
 /**
@@ -42,7 +43,7 @@ public class KafkaSelectedTab extends SelectedTab {
     public List<String> partitionFields;
 
     public static List<Option> getPtCandidateFields() {
-        return SelectedTab.getContextOpts((cols) -> cols.stream());
+        return SelectedTab.getContextOpts(Collection::stream);
     }
 
 

tis-incr/tis-flink-chunjun-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/pipeline/kafka/sink/KafkaPipelineEventSinkFunc.java

@@ -1,21 +1,50 @@
 package com.qlangtech.tis.plugins.incr.flink.cdc.pipeline.kafka.sink;
 
+import com.google.common.collect.Maps;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
 import com.qlangtech.tis.datax.IDataxProcessor;
+import com.qlangtech.tis.datax.IDataxReader;
+import com.qlangtech.tis.plugin.ds.DBConfig;
+import com.qlangtech.tis.plugin.ds.IDataSourceFactoryGetter;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugins.datax.kafka.writer.DataXKafkaWriter;
 import com.qlangtech.tis.plugins.incr.flink.cdc.pipeline.PipelineEventSinkFunc;
 import com.qlangtech.tis.plugins.incr.flink.cdc.pipeline.PipelineFlinkCDCSinkFactory;
+import com.qlangtech.tis.sql.parser.tuple.creator.EntityName;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.factories.Factory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
 import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.connectors.kafka.json.ChangeLogJsonFormatFactory;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSink;
+import org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkFactory;
+import org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkOptions;
+import org.apache.flink.cdc.connectors.kafka.sink.KeyFormat;
+import org.apache.flink.cdc.connectors.kafka.sink.KeySerializationFactory;
+import org.apache.flink.cdc.connectors.kafka.sink.PartitionStrategy;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+import org.apache.kafka.clients.producer.ProducerConfig;
 
 import java.time.ZoneId;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkOptions.KEY_FORMAT;
+import static org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkOptions.PROPERTIES_PREFIX;
+import static org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkOptions.SINK_TABLE_ID_TO_TOPIC_MAPPING;
 
 /**
  * <a href="https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/docs/connectors/pipeline-connectors/kafka/">...</a>
+ *
  * @author 百岁 (baisui@qlangtech.com)
  * @date 2026/3/13
  * // @see tis-flink-pipeline-paimon-plugin:com.qlangtech.tis.plugins.incr.flink.pipeline.paimon.sink.PaimonPipelineEventSinkFunc
@@ -38,37 +67,149 @@ public KafkaPipelineEventSinkFunc(IDataxProcessor dataxProcessor
             , List<ISelectedTab> tabs //
             , IFlinkColCreator<FlinkCol> sourceFlinkColCreator //
             , int sinkTaskParallelism) {
-        super(dataxProcessor, pipelineSinkFactory, sinkDBName, tabs, sourceFlinkColCreator, null, sinkTaskParallelism);
+        super(dataxProcessor, pipelineSinkFactory //
+                , sinkDBName, tabs, sourceFlinkColCreator, null, sinkTaskParallelism);
+    }
+
+    @Override
+    protected EntityName getWriterEntityName(ISelectedTab selTab) {
+
+        IDataxReader reader = this.dataxProcessor.getReader(null);
+        if (reader instanceof IDataSourceFactoryGetter) {
+            try {
+                String[] databaseName = new String[1];
+                DBConfig dbConfig = ((IDataSourceFactoryGetter) reader).getDataSourceFactory().getDbConfig();
+                dbConfig.vistDbName(new DBConfig.IProcess() {
+                    @Override
+                    public boolean visit(DBConfig config, String jdbcUrl, String ip, String dbName) throws Exception {
+                        databaseName[0] = dbName;
+                        return true;
+                    }
+                });
+                return EntityName.create(databaseName[0], selTab.getName());
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        return super.getWriterEntityName(selTab);
     }
 
     /**
      * @param context
      * @return
-     * @see org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSink
+     * @see org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkFactory#createDataSink(Factory.Context)
      */
+//    protected DataSink createDataSink(Factory.Context context) {
+//         KafkaDataSinkFactory factory = new KafkaDataSinkFactory();
+//        return factory.createDataSink(context);
+//    }
     @Override
-    protected DataSink createDataSink(Factory.Context context) {
-//        DeliveryGuarantee deliveryGuarantee,
-//        Properties kafkaProperties,
-//        PartitionStrategy partitionStrategy,
-//        ZoneId zoneId,
-//        SerializationSchema<Event> keySerialization,
-//        SerializationSchema<Event> valueSerialization,
-//        String topic,
-//        boolean addTableToHeaderEnabled,
-//        String customHeaders,
-//        String tableMapping
-
-        String topic = this.writer.topic;
-        ZoneId zoneId = this.pipelineSinkFactory.getTimeZone();
-//        return new KafkaDataSink(catalogOpts, tableOptions, commitUser, partitionMaps, serializer, zoneId,
-//                schemaOperatorUid);
-        return null;
+    public DataSink createDataSink(Factory.Context context) {
+        KafkaPipelineSinkFactory pipelineSinkFactory = ((KafkaPipelineSinkFactory) this.pipelineSinkFactory);
+        KeyFormat keyFormat = context.getFactoryConfiguration().get(KEY_FORMAT);
+        JsonSerializationType jsonSerializationType =
+                context.getFactoryConfiguration().get(KafkaDataSinkOptions.VALUE_FORMAT);
+        KafkaDataSinkFactory factory = new KafkaDataSinkFactory();
+        FactoryHelper helper = FactoryHelper.createFactoryHelper(factory, context);
+        helper.validateExcept(
+                PROPERTIES_PREFIX, keyFormat.toString(), jsonSerializationType.toString());
+
+        DeliveryGuarantee deliveryGuarantee = pipelineSinkFactory.deliveryGuarantee();
+        //   context.getFactoryConfiguration().get(KafkaDataSinkOptions.DELIVERY_GUARANTEE);
+        ZoneId zoneId = ZoneId.systemDefault();
+        if (!Objects.equals(
+                context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+                PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
+            zoneId =
+                    ZoneId.of(
+                            context.getPipelineConfiguration()
+                                    .get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
+        }
+        SerializationSchema<Event> keySerialization =
+                KeySerializationFactory.createSerializationSchema(
+                        helper.getFormatConfig(keyFormat.toString()), keyFormat, zoneId);
+        org.apache.flink.configuration.Configuration formatCfg = pipelineSinkFactory.format.createFlinkCfg();
+
+        SerializationSchema<Event> valueSerialization =
+                ChangeLogJsonFormatFactory.createSerializationSchema(
+                        formatCfg,   // helper.getFormatConfig(jsonSerializationType.toString()),
+                        jsonSerializationType,
+                        zoneId);
+        final Properties kafkaProperties = new Properties();
+        Map<String, String> allOptions = context.getFactoryConfiguration().toMap();
+        allOptions.keySet().stream()
+                .filter(key -> key.startsWith(PROPERTIES_PREFIX))
+                .forEach(
+                        key -> {
+                            final String value = allOptions.get(key);
+                            final String subKey = key.substring((PROPERTIES_PREFIX).length());
+                            kafkaProperties.put(subKey, value);
+                        });
+        String topic = context.getFactoryConfiguration().get(KafkaDataSinkOptions.TOPIC);
+        boolean addTableToHeaderEnabled =
+                context.getFactoryConfiguration()
+                        .get(KafkaDataSinkOptions.SINK_ADD_TABLEID_TO_HEADER_ENABLED);
+        String customHeaders =
+                context.getFactoryConfiguration().get(KafkaDataSinkOptions.SINK_CUSTOM_HEADER);
+        PartitionStrategy partitionStrategy = pipelineSinkFactory.getPartitionStrategy();
+        // context.getFactoryConfiguration().get(KafkaDataSinkOptions.PARTITION_STRATEGY);
+        String tableMapping = context.getFactoryConfiguration().get(SINK_TABLE_ID_TO_TOPIC_MAPPING);
+        return new KafkaDataSink(
+                deliveryGuarantee,
+                kafkaProperties,
+                partitionStrategy,
+                zoneId,
+                keySerialization,
+                valueSerialization,
+                topic,
+                addTableToHeaderEnabled,
+                customHeaders,
+                tableMapping);
     }
 
     @Override
     public Factory.Context createDataSinkContext() {
-        return null;
+        ImmutableMap.Builder<String, String> factoryParamBuilder = ImmutableMap.<String, String>builder();
+
+        // 1. topic from DataXKafkaWriter
+        factoryParamBuilder.put(KafkaDataSinkOptions.TOPIC.key(), this.writer.topic);
+
+        // 2. value format from KafkaPipelineSinkFactory.format
+        KafkaPipelineSinkFactory sinkFactory = (KafkaPipelineSinkFactory) this.pipelineSinkFactory;
+        factoryParamBuilder.put(KafkaDataSinkOptions.VALUE_FORMAT.key(),
+                Objects.requireNonNull(sinkFactory.format, "format can not be null").getDescriptor().getDisplayName());
+
+        // 3. kafka producer properties with "properties." prefix from DataXKafkaWriter
+        Map<String, Object> kafkaConfig = this.writer.buildKafkaConfig();
+        kafkaConfig = Maps.newHashMap(kafkaConfig);
+//        kafkaConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,);
+//        kafkaConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,);
+
+        kafkaConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG
+                , "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArraySerializer");//   .StringSerializer.class.getName())
+        kafkaConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG
+                , "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArraySerializer");//
+
+        for (Map.Entry<String, Object> entry : kafkaConfig.entrySet()) {
+//            if (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG.equals(entry.getKey())
+//                    || ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG.equals(entry.getKey())) {
+//                continue;
+//            }
+            factoryParamBuilder.put(KafkaDataSinkOptions.PROPERTIES_PREFIX + entry.getKey(),
+                    String.valueOf(entry.getValue()));
+        }
+
+        Configuration factoryConfiguration = Configuration.fromMap(factoryParamBuilder.build());
+
+        // 4. pipeline configuration with timezone
+        ImmutableMap.Builder<String, String> pipelineParamBuilder = ImmutableMap.<String, String>builder();
+        pipelineParamBuilder.put(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.key(),
+                this.pipelineSinkFactory.getTimeZone().getId());
+        Configuration pipelineConfiguration = Configuration.fromMap(pipelineParamBuilder.build());
+
+        ClassLoader classLoader = KafkaPipelineEventSinkFunc.class.getClassLoader();
+        return new FactoryHelper.DefaultContext(factoryConfiguration, pipelineConfiguration, classLoader);
     }