You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/04/06 07:35:01 UTC
[incubator-inlong] branch master updated: [INLONG-3537][Manager] Remove unused APIs and change Kafka sink params (#3539)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new d0820a8db [INLONG-3537][Manager] Remove unused APIs and change Kafka sink params (#3539)
d0820a8db is described below
commit d0820a8dbb5a4ea5e26eb1700b0ab1a8f1c553d1
Author: healchow <he...@gmail.com>
AuthorDate: Wed Apr 6 15:34:57 2022 +0800
[INLONG-3537][Manager] Remove unused APIs and change Kafka sink params (#3539)
---
.../inlong/manager/client/Binlog2KafkaExample.java | 2 +-
.../inlong/manager/client/api/sink/KafkaSink.java | 2 +-
.../client/api/util/InlongStreamSinkTransfer.java | 6 +++---
.../apache/inlong/manager/common/enums/Constant.java | 2 +-
.../manager/common/pojo/sink/kafka/KafkaSinkDTO.java | 13 ++++++++++---
.../common/pojo/sink/kafka/KafkaSinkListResponse.java | 9 ++++++++-
.../common/pojo/sink/kafka/KafkaSinkRequest.java | 10 +++++++++-
.../common/pojo/sink/kafka/KafkaSinkResponse.java | 17 ++++++++++++-----
.../manager/service/core/InlongStreamService.java | 11 -----------
.../service/core/impl/InlongStreamServiceImpl.java | 11 -----------
.../service/thirdparty/sort/util/SinkInfoUtils.java | 2 +-
.../service/core/sink/KafkaStreamSinkServiceTest.java | 2 +-
.../manager/web/controller/InlongStreamController.java | 8 --------
13 files changed, 47 insertions(+), 48 deletions(-)
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
index cbf259264..eccb9c618 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
@@ -199,7 +199,7 @@ public class Binlog2KafkaExample {
private KafkaSink createKafkaSink() {
KafkaSink kafkaSink = new KafkaSink();
kafkaSink.setDataFormat(DataFormat.CANAL);
- kafkaSink.setAddress("{kafka.bootstrap}");
+ kafkaSink.setBootstrapServers("{kafka.bootstrap}");
kafkaSink.setTopicName("{kafka.topic}");
kafkaSink.setNeedCreated(false);
kafkaSink.setSinkName("{kafka.sink.name}");
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/KafkaSink.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/KafkaSink.java
index 801d2d830..5b5e97288 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/KafkaSink.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/KafkaSink.java
@@ -41,7 +41,7 @@ public class KafkaSink extends StreamSink {
private SinkType sinkType = SinkType.KAFKA;
@ApiModelProperty("Kafka bootstrap servers")
- private String address;
+ private String bootstrapServers;
@ApiModelProperty("Kafka topicName")
private String topicName;
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
index 751469de7..07c7bd051 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
@@ -164,7 +164,7 @@ public class InlongStreamSinkTransfer {
KafkaSinkRequest kafkaSinkRequest = new KafkaSinkRequest();
KafkaSink kafkaSink = (KafkaSink) streamSink;
kafkaSinkRequest.setSinkName(streamSink.getSinkName());
- kafkaSinkRequest.setAddress(kafkaSink.getAddress());
+ kafkaSinkRequest.setBootstrapServers(kafkaSink.getBootstrapServers());
kafkaSinkRequest.setTopicName(kafkaSink.getTopicName());
kafkaSinkRequest.setSinkType(kafkaSink.getSinkType().name());
kafkaSinkRequest.setInlongGroupId(streamInfo.getInlongGroupId());
@@ -186,12 +186,12 @@ public class InlongStreamSinkTransfer {
String.format("SinkName is not equal: %s != %s", sinkResponse, sink));
KafkaSink snapshot = (KafkaSink) sink;
kafkaSink.setSinkName(snapshot.getSinkName());
- kafkaSink.setAddress(snapshot.getAddress());
+ kafkaSink.setBootstrapServers(snapshot.getBootstrapServers());
kafkaSink.setTopicName(snapshot.getTopicName());
kafkaSink.setDataFormat(snapshot.getDataFormat());
} else {
kafkaSink.setSinkName(sinkResponse.getSinkName());
- kafkaSink.setAddress(sinkResponse.getAddress());
+ kafkaSink.setBootstrapServers(sinkResponse.getBootstrapServers());
kafkaSink.setTopicName(sinkResponse.getTopicName());
kafkaSink.setDataFormat(DataFormat.forName(sinkResponse.getSerializationType()));
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
index 643a0cc7b..f614460d9 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
@@ -71,7 +71,7 @@ public class Constant {
public static final String ID_IS_EMPTY = "primary key is empty";
- public static final String GROUP_ID_IS_EMPTY = "data group id is empty";
+ public static final String GROUP_ID_IS_EMPTY = "inlong group id is empty";
public static final String STREAM_ID_IS_EMPTY = "inlong stream id is empty";
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkDTO.java
index 54b4fc605..1f7f853d5 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkDTO.java
@@ -20,7 +20,6 @@ package org.apache.inlong.manager.common.pojo.sink.kafka;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.annotations.ApiModelProperty;
-import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -29,6 +28,7 @@ import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.exceptions.BusinessException;
import javax.validation.constraints.NotNull;
+import java.util.Map;
/**
* Kafka sink info
@@ -42,14 +42,21 @@ public class KafkaSinkDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@ApiModelProperty("Kafka bootstrap servers")
- private String address;
+ private String bootstrapServers;
@ApiModelProperty("Kafka topicName")
private String topicName;
+ @ApiModelProperty("Partition number of the topic")
+ private String partitionNum;
+
@ApiModelProperty("Data Serialization, support: json, canal, avro")
private String serializationType;
+ @ApiModelProperty(value = "The strategy of auto offset reset",
+ notes = "including earliest, latest (the default), none")
+ private String autoOffsetReset;
+
@ApiModelProperty("Properties for kafka")
private Map<String, Object> properties;
@@ -58,7 +65,7 @@ public class KafkaSinkDTO {
*/
public static KafkaSinkDTO getFromRequest(KafkaSinkRequest request) {
return KafkaSinkDTO.builder()
- .address(request.getAddress())
+ .bootstrapServers(request.getBootstrapServers())
.topicName(request.getTopicName())
.serializationType(request.getSerializationType())
.properties(request.getProperties())
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkListResponse.java
index eff2b46d8..85c4515ec 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkListResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkListResponse.java
@@ -32,12 +32,19 @@ import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
public class KafkaSinkListResponse extends SinkListResponse {
@ApiModelProperty("Kafka bootstrap servers")
- private String address;
+ private String bootstrapServers;
@ApiModelProperty("Kafka topicName")
private String topicName;
+ @ApiModelProperty("Partition number of the topic")
+ private String partitionNum;
+
@ApiModelProperty("Data Serialization, support: json, canal, avro")
private String serializationType;
+ @ApiModelProperty(value = "The strategy of auto offset reset",
+ notes = "including earliest, latest (the default), none")
+ private String autoOffsetReset;
+
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkRequest.java
index 74ca8e475..791f8dc08 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkRequest.java
@@ -37,11 +37,19 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
public class KafkaSinkRequest extends SinkRequest {
@ApiModelProperty("Kafka bootstrap servers")
- private String address;
+ private String bootstrapServers;
@ApiModelProperty("Kafka topicName")
private String topicName;
+ @ApiModelProperty("Partition number of the topic")
+ private String partitionNum;
+
@ApiModelProperty("Data Serialization, support: json, canal, avro")
private String serializationType;
+
+ @ApiModelProperty(value = "The strategy of auto offset reset",
+ notes = "including earliest, latest (the default), none")
+ private String autoOffsetReset;
+
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkResponse.java
index b60e2b18b..f7929445f 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/kafka/KafkaSinkResponse.java
@@ -34,17 +34,24 @@ import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
@ApiModel(value = "Response of the Kafka sink")
public class KafkaSinkResponse extends SinkResponse {
- public KafkaSinkResponse() {
- this.sinkType = Constant.SINK_KAFKA;
- }
-
@ApiModelProperty("Kafka bootstrap servers")
- private String address;
+ private String bootstrapServers;
@ApiModelProperty("Kafka topicName")
private String topicName;
+ @ApiModelProperty("Partition number of the topic")
+ private String partitionNum;
+
@ApiModelProperty("Data Serialization, support: json, canal, avro")
private String serializationType;
+ @ApiModelProperty(value = "The strategy of auto offset reset",
+ notes = "including earliest, latest (the default), none")
+ private String autoOffsetReset;
+
+ public KafkaSinkResponse() {
+ this.sinkType = Constant.SINK_KAFKA;
+ }
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
index 0487dc12d..ed84b8788 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
@@ -137,17 +137,6 @@ public interface InlongStreamService {
*/
PageInfo<FullStreamResponse> listAllWithGroupId(InlongStreamPageRequest request);
- /**
- * Modify all inlong streams (including basic information about data sources)
- *
- * @param updateInfo inlong stream page information
- * @param operator Edit person's name
- * @return Whether the modification is successful
- * @apiNote The data source details and data sink information are modified separately,
- * not in this all modification interface
- */
- boolean updateAll(InlongStreamRequest updateInfo, String operator);
-
/**
* According to the group id, query the number of valid inlong streams belonging to this service
*
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
index f66cd150f..d0a31d5b1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
@@ -445,17 +445,6 @@ public class InlongStreamServiceImpl implements InlongStreamService {
return pageInfo;
}
- @Transactional(rollbackFor = Throwable.class)
- @Override
- public boolean updateAll(InlongStreamRequest request, String operator) {
- LOGGER.info("begin to update all stream page info: " + request);
- Preconditions.checkNotNull(request, "request is empty");
-
- this.update(request, operator);
- LOGGER.info("success to update all stream page info");
- return true;
- }
-
@Override
public int selectCountByGroupId(String groupId) {
LOGGER.debug("begin to get count by groupId={}", groupId);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SinkInfoUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SinkInfoUtils.java
index 13bf46cfe..ad9a734d5 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SinkInfoUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SinkInfoUtils.java
@@ -110,7 +110,7 @@ public class SinkInfoUtils {
private static KafkaSinkInfo createKafkaSinkInfo(SourceResponse sourceResponse, KafkaSinkResponse sinkResponse,
List<FieldInfo> sinkFields) {
- String addressUrl = sinkResponse.getAddress();
+ String addressUrl = sinkResponse.getBootstrapServers();
String topicName = sinkResponse.getTopicName();
SerializationInfo serializationInfo = SerializationUtils.createSerialInfo(sourceResponse,
sinkResponse);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/KafkaStreamSinkServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/KafkaStreamSinkServiceTest.java
index e5ba8f6fb..9d441349f 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/KafkaStreamSinkServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/KafkaStreamSinkServiceTest.java
@@ -59,7 +59,7 @@ public class KafkaStreamSinkServiceTest extends ServiceBaseTest {
sinkInfo.setSinkType(Constant.SINK_KAFKA);
sinkInfo.setSinkName(sinkName);
sinkInfo.setSerializationType(serializationType);
- sinkInfo.setAddress(bootstrapServers);
+ sinkInfo.setBootstrapServers(bootstrapServers);
sinkInfo.setTopicName(topicName);
sinkInfo.setEnableCreateResource(Constant.DISABLE_CREATE_RESOURCE);
kafkaSinkId = sinkService.save(sinkInfo, globalOperator);
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
index 430dd2a65..f8b5711a3 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
@@ -110,14 +110,6 @@ public class InlongStreamController {
return Response.success(streamService.update(request, username));
}
- @RequestMapping(value = "/updateAll", method = RequestMethod.POST)
- @OperationLog(operation = OperationType.UPDATE)
- @ApiOperation(value = "Update inlong stream page info")
- public Response<Boolean> updateAll(@RequestBody InlongStreamRequest request) {
- boolean result = streamService.updateAll(request, LoginUserUtils.getLoginUserDetail().getUserName());
- return Response.success(result);
- }
-
@RequestMapping(value = "/delete", method = RequestMethod.DELETE)
@OperationLog(operation = OperationType.DELETE)
@ApiOperation(value = "Delete inlong stream info")