feat: TAP-6051 DB - Kafka Transformer

Closes TAP-6051

Author: ply0011

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/schema_mode/OriginalSchemaMode.java

@@ -68,10 +68,10 @@ public TapEvent toTapEvent(ConsumerRecord<?, ?> consumerRecord) {
     public List<ProducerRecord<Object, Object>> fromTapEvent(TapTable table, TapEvent tapEvent) {
         if (tapEvent instanceof TapInsertRecordEvent) {
             TapInsertRecordEvent insertRecordEvent = (TapInsertRecordEvent) tapEvent;
-            String topic = KafkaUtils.pickTopic(kafkaService.getConfig(), "", "", table);
+            String topic = topic(table, tapEvent);
             Map<String, Object> afterMap = insertRecordEvent.getAfter();
             Object key = afterMap.get(FIELD_KEY);
-            Object value = afterMap.get(FIELD_VALUE);
+            Object value = afterMap.containsKey(FIELD_KEY) ? afterMap.get(FIELD_VALUE) : afterMap;
             Integer partition = (Integer) afterMap.get(FIELD_PARTITION);
             Long ts = Optional.ofNullable(afterMap.get(FIELD_TIMESTAMP)).map(o -> {
                 if (o instanceof String) {

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/schema_mode/StandardSchemaMode.java

@@ -77,7 +77,7 @@ public TapEvent toTapEvent(ConsumerRecord<?, ?> consumerRecord) {
 
     @Override
     public List<ProducerRecord<Object, Object>> fromTapEvent(TapTable table, TapEvent tapEvent) {
-        String topic = KafkaUtils.pickTopic(kafkaService.getConfig(), "", "", table);
+        String topic = topic(table, tapEvent);
         Long ts = tapEvent.getTime();
 
         Map<String, Object> data;

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/serialization/StandardSerializer.java

@@ -8,8 +8,7 @@
 import io.tapdata.kafka.utils.StandardEventUtils;
 import org.apache.kafka.common.serialization.Serializer;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 
 /**
  * TapData 标准事件序列器
@@ -25,9 +24,9 @@ public byte[] serialize(String topic, TapEvent data) {
         StandardEventUtils.setTs(map, data.getTime());
         if (data instanceof TapBaseEvent) {
             TapBaseEvent recordEvent = (TapBaseEvent) data;
-            StandardEventUtils.setOpTs(map, recordEvent.getReferenceTime());
+            StandardEventUtils.setOpTs(map, null != recordEvent.getReferenceTime() ? recordEvent.getReferenceTime() : System.currentTimeMillis());
             StandardEventUtils.setTable(map, recordEvent.getTableId());
-            StandardEventUtils.setNamespaces(map, recordEvent.getNamespaces());
+            StandardEventUtils.setNamespaces(map, getNamespaces(recordEvent));
         } else {
             StandardEventUtils.setTable(map, topic);
         }
@@ -75,4 +74,27 @@ private void toUnknownMap(Map<String, Object> map, TapEvent recordEvent) {
         StandardEventUtils.setOp(map, EventOperation.DML_UNKNOWN);
         StandardEventUtils.setData(map, recordEvent);
     }
+
+    private List<String> getNamespaces(TapBaseEvent tapBaseEvent) {
+        if (null == tapBaseEvent) {
+            return null;
+        }
+        List<String> namespaces = tapBaseEvent.getNamespaces();
+        if (null != namespaces) {
+            return namespaces;
+        }
+        List<String> ns = new ArrayList<>();
+        String database = tapBaseEvent.getDatabase();
+        if (null != database) {
+            ns.add(database);
+        }
+        String schema = tapBaseEvent.getSchema();
+        if (null != schema) {
+            ns.add(schema);
+        }
+        if (!ns.isEmpty()) {
+            return ns;
+        }
+        return null;
+    }
 }