[Manager] Fix sort standalone get kafka config error (#10106)

Co-authored-by: castorqin <qhj00725@qq.com>
apache · May 4, 2024 · 2e70a12 · 2e70a12
1 parent 6260e1a
commit 2e70a12
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 0 deletions.
diff --git a/...ice/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java b/...ice/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java
@@ -40,6 +40,8 @@
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
 
+import java.util.Map;
+
 /**
  * Kafka data node operator
  */
@@ -48,6 +50,9 @@ public class KafkaDataNodeOperator extends AbstractDataNodeOperator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaDataNodeOperator.class);
 
+    private static final String bootstrapServers = "bootstrap.servers";
+    private static final String clientId = "client.id";
+
     @Autowired
     private ObjectMapper objectMapper;
 
@@ -79,6 +84,15 @@ public DataNodeInfo getFromEntity(DataNodeEntity entity) {
         return kafkaDataNodeInfo;
     }
 
+    @Override
+    public Map<String, String> parse2SinkParams(DataNodeInfo info) {
+        Map<String, String> params = super.parse2SinkParams(info);
+        KafkaDataNodeInfo kafkaDataNodeInfo = (KafkaDataNodeInfo) info;
+        params.put(bootstrapServers, kafkaDataNodeInfo.getBootstrapServers());
+        params.put(clientId, kafkaDataNodeInfo.getClientId());
+        return params;
+    }
+
     @Override
     protected void setTargetEntity(DataNodeRequest request, DataNodeEntity targetEntity) {
         KafkaDataNodeRequest nodeRequest = (KafkaDataNodeRequest) request;

diff --git a/...service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java b/...service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java
@@ -22,6 +22,7 @@
 import org.apache.inlong.manager.common.exceptions.BusinessException;
 import org.apache.inlong.manager.common.util.CommonBeanUtils;
 import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
+import org.apache.inlong.manager.pojo.node.DataNodeInfo;
 import org.apache.inlong.manager.pojo.sink.SinkField;
 import org.apache.inlong.manager.pojo.sink.SinkRequest;
 import org.apache.inlong.manager.pojo.sink.StreamSink;
@@ -30,13 +31,15 @@
 import org.apache.inlong.manager.pojo.sink.kafka.KafkaSinkRequest;
 import org.apache.inlong.manager.service.sink.AbstractSinkOperator;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * Kafka sink operator
@@ -46,6 +49,8 @@ public class KafkaSinkOperator extends AbstractSinkOperator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSinkOperator.class);
 
+    private static final String topic = "topic";
+
     @Autowired
     private ObjectMapper objectMapper;
 
@@ -75,6 +80,23 @@ protected void setTargetEntity(SinkRequest request, StreamSinkEntity targetEntit
         }
     }
 
+    @Override
+    public Map<String, String> parse2IdParams(StreamSinkEntity streamSink, List<String> fields,
+            DataNodeInfo dataNodeInfo) {
+
+        Map<String, String> params = super.parse2IdParams(streamSink, fields, dataNodeInfo);
+
+        KafkaSinkDTO kafkaSinkDTO;
+        try {
+            kafkaSinkDTO = objectMapper.readValue(streamSink.getExtParams(), KafkaSinkDTO.class);
+        } catch (JsonProcessingException e) {
+            LOGGER.error("parse kafka sink dto error", e);
+            return params;
+        }
+        params.put(topic, kafkaSinkDTO.getTopicName());
+        return params;
+    }
+
     @Override
     public StreamSink getFromEntity(StreamSinkEntity entity) {
         KafkaSink sink = new KafkaSink();