You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2024/05/04 06:25:50 UTC
(inlong) branch revert-10106-INLONG-10105 created (now 7aeb30664c)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a change to branch revert-10106-INLONG-10105
in repository https://gitbox.apache.org/repos/asf/inlong.git
at 7aeb30664c Revert "[Manager] Fix sort standalone get kafka config error (#10106)"
This branch includes the following new commits:
new 7aeb30664c Revert "[Manager] Fix sort standalone get kafka config error (#10106)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(inlong) 01/01: Revert "[Manager] Fix sort standalone get kafka config error (#10106)"
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch revert-10106-INLONG-10105
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 7aeb30664ca9a57460725a6d0f3e07183ca6f07f
Author: Charles Zhang <do...@apache.org>
AuthorDate: Sat May 4 14:25:45 2024 +0800
Revert "[Manager] Fix sort standalone get kafka config error (#10106)"
This reverts commit 2e70a12a2f95e9c8eacdc6ce91afa6f4297e2df3.
---
.../service/node/kafka/KafkaDataNodeOperator.java | 14 --------------
.../service/sink/kafka/KafkaSinkOperator.java | 22 ----------------------
2 files changed, 36 deletions(-)
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java
index 4fbc740d36..ae91b2f394 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/kafka/KafkaDataNodeOperator.java
@@ -40,8 +40,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
-import java.util.Map;
-
/**
* Kafka data node operator
*/
@@ -50,9 +48,6 @@ 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;
@@ -84,15 +79,6 @@ public class KafkaDataNodeOperator extends AbstractDataNodeOperator {
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/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java
index d7fa197c9f..4357556732 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/kafka/KafkaSinkOperator.java
@@ -22,7 +22,6 @@ import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
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;
@@ -31,7 +30,6 @@ import org.apache.inlong.manager.pojo.sink.kafka.KafkaSinkDTO;
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;
@@ -39,7 +37,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
-import java.util.Map;
/**
* Kafka sink operator
@@ -49,8 +46,6 @@ public class KafkaSinkOperator extends AbstractSinkOperator {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSinkOperator.class);
- private static final String topic = "topic";
-
@Autowired
private ObjectMapper objectMapper;
@@ -80,23 +75,6 @@ public class KafkaSinkOperator extends AbstractSinkOperator {
}
}
- @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();