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/07/12 03:01:53 UTC
[inlong] branch master updated: [INLONG-4993][Manager] Return details when querying a list of StreamSources (#4994)
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/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 3781c8292 [INLONG-4993][Manager] Return details when querying a list of StreamSources (#4994)
3781c8292 is described below
commit 3781c8292517de9fb9955cd1af6d3266c78de81b
Author: healchow <he...@gmail.com>
AuthorDate: Tue Jul 12 11:01:48 2022 +0800
[INLONG-4993][Manager] Return details when querying a list of StreamSources (#4994)
---
.../inlong/manager/client/cli/DescribeCommand.java | 6 +-
.../inlong/manager/client/cli/ListCommand.java | 6 +-
.../api/impl/DefaultInlongStreamBuilder.java | 13 ++-
.../manager/client/api/impl/InlongClientImpl.java | 13 ++-
.../manager/client/api/impl/InlongStreamImpl.java | 13 ++-
.../client/api/inner/InnerInlongManagerClient.java | 8 +-
.../client/api/service/StreamSourceApi.java | 4 +-
.../api/inner/InnerInlongManagerClientTest.java | 44 ++++-----
.../manager/common/pojo/cluster/ClusterInfo.java | 7 +-
.../common/pojo/cluster/ClusterRequest.java | 4 +-
.../pojo/consumption/ConsumptionMqExtBase.java | 4 +-
.../manager/common/pojo/group/InlongGroupInfo.java | 3 +-
.../common/pojo/group/InlongGroupListResponse.java | 4 +-
.../common/pojo/group/InlongGroupRequest.java | 10 +-
.../manager/common/pojo/sink/SinkListResponse.java | 5 +-
.../manager/common/pojo/sink/SinkRequest.java | 5 +-
.../manager/common/pojo/sink/StreamSink.java | 3 +-
.../manager/common/pojo/sort/BaseSortConf.java | 3 +-
.../common/pojo/source/SourceListResponse.java | 77 ----------------
.../manager/common/pojo/source/SourceRequest.java | 3 +-
.../manager/common/pojo/source/StreamSource.java | 3 +-
.../autopush/AutoPushSourceListResponse.java | 47 ----------
.../pojo/source/file/FileSourceListResponse.java | 57 ------------
.../common/pojo/source/kafka/KafkaSource.java | 6 +-
.../pojo/source/kafka/KafkaSourceListResponse.java | 71 ---------------
.../common/pojo/source/mongodb/MongoDBSource.java | 16 ++--
.../pojo/source/mongodb/MongoDBSourceDTO.java | 16 ++--
.../source/mongodb/MongoDBSourceListResponse.java | 63 -------------
.../pojo/source/mongodb/MongoDBSourceRequest.java | 16 ++--
.../pojo/source/mysql/MySQLBinlogSource.java | 16 ++--
.../pojo/source/mysql/MySQLBinlogSourceDTO.java | 16 ++--
.../mysql/MySQLBinlogSourceListResponse.java | 101 ---------------------
.../source/mysql/MySQLBinlogSourceRequest.java | 8 +-
.../common/pojo/source/oracle/OracleSource.java | 14 +--
.../common/pojo/source/oracle/OracleSourceDTO.java | 14 +--
.../source/oracle/OracleSourceListResponse.java | 75 ---------------
.../pojo/source/oracle/OracleSourceRequest.java | 14 +--
.../postgres/PostgresSourceListResponse.java | 73 ---------------
.../PostgreSQLSource.java} | 34 +++----
.../PostgreSQLSourceDTO.java} | 36 ++++----
.../PostgreSQLSourceRequest.java} | 32 +++----
.../source/pulsar/PulsarSourceListResponse.java | 66 --------------
.../{SqlServerSource.java => SQLServerSource.java} | 28 +++---
...erverSourceDTO.java => SQLServerSourceDTO.java} | 30 +++---
...rceRequest.java => SQLServerSourceRequest.java} | 22 ++---
.../sqlserver/SqlServerSourceListResponse.java | 75 ---------------
.../service/group/InlongGroupServiceImpl.java | 14 ++-
.../service/sort/util/ExtractNodeUtils.java | 30 +++---
.../service/source/AbstractSourceOperator.java | 33 +++----
.../service/source/StreamSourceOperator.java | 19 ++--
.../service/source/StreamSourceService.java | 7 +-
.../service/source/StreamSourceServiceImpl.java | 13 ++-
.../source/autopush/AutoPushSourceOperator.java | 52 ++++-------
.../source/binlog/BinlogSourceOperator.java | 38 +++-----
.../service/source/file/FileSourceOperator.java | 55 +++++------
.../service/source/kafka/KafkaSourceOperator.java | 40 +++-----
.../{mongo => mongodb}/MongoDBSourceOperator.java | 42 +++------
.../source/oracle/OracleSourceOperator.java | 38 +++-----
.../PostgreSQLSourceOperator.java} | 56 +++++-------
.../source/pulsar/PulsarSourceOperator.java | 56 +++++-------
...eOperator.java => SQLServerSourceOperator.java} | 53 ++++-------
.../source/MongoDBSourceServiceTest.java | 3 +-
.../{core => }/source/OracleSourceServiceTest.java | 3 +-
.../SQLServerSourceServiceTest.java} | 15 ++-
.../{core => }/source/StreamSourceServiceTest.java | 3 +-
.../web/controller/StreamSourceController.java | 3 +-
.../org/apache/inlong/manager/web/UtilsTest.java | 5 +-
67 files changed, 452 insertions(+), 1310 deletions(-)
diff --git a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/DescribeCommand.java b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/DescribeCommand.java
index f8032cc6d..eb6d1560b 100644
--- a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/DescribeCommand.java
+++ b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/DescribeCommand.java
@@ -28,7 +28,7 @@ import org.apache.inlong.manager.client.cli.util.PrintUtils;
import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.io.IOException;
@@ -174,8 +174,8 @@ public class DescribeCommand extends AbstractCommand {
@Override
void run() {
try {
- List<SourceListResponse> sourceListResponses = managerClient.listSources(group, stream, type);
- sourceListResponses.forEach(PrintUtils::printJson);
+ List<StreamSource> sources = managerClient.listSources(group, stream, type);
+ sources.forEach(PrintUtils::printJson);
} catch (Exception e) {
System.out.println(e.getMessage());
}
diff --git a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/ListCommand.java b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/ListCommand.java
index def185ff7..13158f0f7 100644
--- a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/ListCommand.java
+++ b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/ListCommand.java
@@ -32,7 +32,7 @@ import org.apache.inlong.manager.client.cli.util.PrintUtils;
import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.io.IOException;
@@ -180,7 +180,7 @@ public class ListCommand extends AbstractCommand {
@Parameter(names = {"-g", "--group"}, required = true, description = "inlong group id")
private String group;
- @Parameter(names = {"-t", "--type"}, description = "sink type")
+ @Parameter(names = {"-t", "--type"}, description = "source type")
private String type;
ListSource(InnerInlongManagerClient managerClient) {
@@ -190,7 +190,7 @@ public class ListCommand extends AbstractCommand {
@Override
void run() {
try {
- List<SourceListResponse> sourceListResponses = managerClient.listSources(group, stream, type);
+ List<StreamSource> sourceListResponses = managerClient.listSources(group, stream, type);
PrintUtils.print(sourceListResponses, SourceInfo.class);
} catch (Exception e) {
System.out.println(e.getMessage());
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
index 22c400943..d26a52f1d 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
@@ -32,7 +32,6 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
import org.apache.inlong.manager.common.pojo.sink.SinkRequest;
import org.apache.inlong.manager.common.pojo.sink.StreamSink;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
@@ -202,12 +201,12 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
InlongStreamInfo streamInfo = streamContext.getStreamInfo();
final String groupId = streamInfo.getInlongGroupId();
final String streamId = streamInfo.getInlongStreamId();
- List<SourceListResponse> sourceListResponses = managerClient.listSources(groupId, streamId);
+ List<StreamSource> streamSources = managerClient.listSources(groupId, streamId);
List<String> updateSourceNames = Lists.newArrayList();
- if (CollectionUtils.isNotEmpty(sourceListResponses)) {
- for (SourceListResponse sourceListResponse : sourceListResponses) {
- final String sourceName = sourceListResponse.getSourceName();
- final int id = sourceListResponse.getId();
+ if (CollectionUtils.isNotEmpty(streamSources)) {
+ for (StreamSource source : streamSources) {
+ final String sourceName = source.getSourceName();
+ final int id = source.getId();
if (sourceRequests.get(sourceName) == null) {
boolean isDelete = managerClient.deleteSource(id);
if (!isDelete) {
@@ -222,7 +221,7 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
updateState.getValue()));
}
updateSourceNames.add(sourceName);
- sourceRequest.setId(sourceListResponse.getId());
+ sourceRequest.setId(source.getId());
}
}
}
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongClientImpl.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongClientImpl.java
index b94475794..0a2c52518 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongClientImpl.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongClientImpl.java
@@ -34,7 +34,7 @@ import org.apache.inlong.manager.client.api.inner.InnerInlongManagerClient;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.util.HttpUtils;
import java.util.List;
@@ -111,8 +111,8 @@ public class InlongClientImpl implements InlongClient {
groupListResponses.forEach(response -> {
String groupId = response.getInlongGroupId();
SimpleGroupStatus groupStatus = SimpleGroupStatus.parseStatusByCode(response.getStatus());
- List<SourceListResponse> sourceListResponses = response.getSourceResponses();
- groupStatus = recheckGroupStatus(groupStatus, sourceListResponses);
+ List<StreamSource> sources = response.getStreamSources();
+ groupStatus = recheckGroupStatus(groupStatus, sources);
groupStatusMap.put(groupId, groupStatus);
});
}
@@ -129,10 +129,9 @@ public class InlongClientImpl implements InlongClient {
return new InlongGroupImpl(groupInfo, this);
}
- private SimpleGroupStatus recheckGroupStatus(SimpleGroupStatus groupStatus,
- List<SourceListResponse> sourceListResponses) {
- Map<SimpleSourceStatus, List<SourceListResponse>> statusListMap = Maps.newHashMap();
- sourceListResponses.forEach(source -> {
+ private SimpleGroupStatus recheckGroupStatus(SimpleGroupStatus groupStatus, List<StreamSource> sources) {
+ Map<SimpleSourceStatus, List<StreamSource>> statusListMap = Maps.newHashMap();
+ sources.forEach(source -> {
SimpleSourceStatus status = SimpleSourceStatus.parseByStatus(source.getStatus());
statusListMap.computeIfAbsent(status, k -> Lists.newArrayList()).add(source);
});
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
index 888127387..537b2db32 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
@@ -29,7 +29,6 @@ import org.apache.inlong.manager.client.api.inner.InnerInlongManagerClient;
import org.apache.inlong.manager.client.api.util.StreamTransformTransfer;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
import org.apache.inlong.manager.common.pojo.sink.StreamSink;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.StreamField;
@@ -341,15 +340,15 @@ public class InlongStreamImpl implements InlongStream {
}
private void initOrUpdateSource(InlongStreamInfo streamInfo) {
- List<SourceListResponse> sourceListResponses = managerClient.listSources(inlongGroupId, inlongStreamId);
+ List<StreamSource> streamSources = managerClient.listSources(inlongGroupId, inlongStreamId);
List<String> updateSourceNames = Lists.newArrayList();
- for (SourceListResponse sourceListResponse : sourceListResponses) {
- final String sourceName = sourceListResponse.getSourceName();
- final int id = sourceListResponse.getId();
+ for (StreamSource source : streamSources) {
+ final String sourceName = source.getSourceName();
+ final int id = source.getId();
if (this.streamSources.get(sourceName) == null) {
boolean isDelete = managerClient.deleteSource(id);
if (!isDelete) {
- throw new RuntimeException(String.format("Delete source=%s failed", sourceListResponse));
+ throw new RuntimeException(String.format("Delete source=%s failed", source));
}
} else {
StreamSource streamSource = this.streamSources.get(sourceName);
@@ -364,7 +363,7 @@ public class InlongStreamImpl implements InlongStream {
updateSourceNames.add(sourceName);
}
}
- for (Map.Entry<String, StreamSource> sourceEntry : streamSources.entrySet()) {
+ for (Map.Entry<String, StreamSource> sourceEntry : this.streamSources.entrySet()) {
String sourceName = sourceEntry.getKey();
if (updateSourceNames.contains(sourceName)) {
continue;
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
index 32f3a1008..71d3a5652 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
@@ -48,8 +48,8 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupRequest;
import org.apache.inlong.manager.common.pojo.group.InlongGroupResetRequest;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
import org.apache.inlong.manager.common.pojo.sink.SinkRequest;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamConfigLogListResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest;
@@ -326,15 +326,15 @@ public class InnerInlongManagerClient {
/**
* Get information of sources.
*/
- public List<SourceListResponse> listSources(String groupId, String streamId) {
+ public List<StreamSource> listSources(String groupId, String streamId) {
return listSources(groupId, streamId, null);
}
/**
* List information of sources by the specified source type.
*/
- public List<SourceListResponse> listSources(String groupId, String streamId, String sourceType) {
- Response<PageInfo<SourceListResponse>> response = executeHttpCall(
+ public List<StreamSource> listSources(String groupId, String streamId, String sourceType) {
+ Response<PageInfo<StreamSource>> response = executeHttpCall(
streamSourceApi.listSources(groupId, streamId, sourceType));
assertRespSuccess(response);
return response.getData().getList();
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
index 0c2ba992b..0f8e0e8d4 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
@@ -19,8 +19,8 @@ package org.apache.inlong.manager.client.api.service;
import com.github.pagehelper.PageInfo;
import org.apache.inlong.manager.common.beans.Response;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import retrofit2.Call;
import retrofit2.http.Body;
import retrofit2.http.DELETE;
@@ -38,7 +38,7 @@ public interface StreamSourceApi {
Call<Response<Boolean>> updateSource(@Body SourceRequest request);
@GET("source/list")
- Call<Response<PageInfo<SourceListResponse>>> listSources(@Query("inlongGroupId") String groupId,
+ Call<Response<PageInfo<StreamSource>>> listSources(@Query("inlongGroupId") String groupId,
@Query("inlongStreamId") String streamId, @Query("sourceType") String sourceType);
@DELETE("source/delete/{id}")
diff --git a/inlong-manager/manager-client/src/test/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClientTest.java b/inlong-manager/manager-client/src/test/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClientTest.java
index 2295c4eb8..4ba24cc22 100644
--- a/inlong-manager/manager-client/src/test/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClientTest.java
+++ b/inlong-manager/manager-client/src/test/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClientTest.java
@@ -29,12 +29,13 @@ import org.apache.inlong.manager.client.api.impl.InlongClientImpl;
import org.apache.inlong.manager.common.auth.DefaultAuthentication;
import org.apache.inlong.manager.common.beans.Response;
import org.apache.inlong.manager.common.pojo.cluster.ClusterRequest;
+import org.apache.inlong.manager.common.pojo.cluster.pulsar.PulsarClusterRequest;
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
-import org.apache.inlong.manager.common.pojo.group.InlongGroupRequest;
import org.apache.inlong.manager.common.pojo.group.InlongGroupResetRequest;
import org.apache.inlong.manager.common.pojo.group.pulsar.InlongPulsarInfo;
+import org.apache.inlong.manager.common.pojo.group.pulsar.InlongPulsarRequest;
import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
import org.apache.inlong.manager.common.pojo.sink.StreamSink;
import org.apache.inlong.manager.common.pojo.sink.ck.ClickHouseSink;
@@ -48,16 +49,11 @@ import org.apache.inlong.manager.common.pojo.sink.iceberg.IcebergSinkListRespons
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSink;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkListResponse;
import org.apache.inlong.manager.common.pojo.sink.postgres.PostgresSinkListResponse;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSource;
-import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.file.FileSource;
-import org.apache.inlong.manager.common.pojo.source.file.FileSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSource;
-import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSource;
-import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSourceListResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
import org.apache.inlong.manager.common.pojo.stream.StreamField;
@@ -154,9 +150,9 @@ class InnerInlongManagerClientTest {
.id(1)
.inlongGroupId("1")
.name("name")
- .sourceResponses(
+ .streamSources(
Lists.newArrayList(
- AutoPushSourceListResponse.builder()
+ AutoPushSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -186,9 +182,9 @@ class InnerInlongManagerClientTest {
.id(1)
.inlongGroupId("1")
.name("name")
- .sourceResponses(
+ .streamSources(
Lists.newArrayList(
- MySQLBinlogSourceListResponse.builder()
+ MySQLBinlogSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -226,9 +222,9 @@ class InnerInlongManagerClientTest {
.status(1)
.createTime(new Date())
.modifyTime(new Date())
- .sourceResponses(
+ .streamSources(
Lists.newArrayList(
- FileSourceListResponse.builder()
+ FileSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -259,9 +255,9 @@ class InnerInlongManagerClientTest {
InlongGroupListResponse.builder()
.id(1)
.inlongGroupId("1")
- .sourceResponses(
+ .streamSources(
Lists.newArrayList(
- KafkaSourceListResponse.builder()
+ KafkaSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -295,8 +291,8 @@ class InnerInlongManagerClientTest {
@Test
void testListGroup4AllSource() {
- ArrayList<SourceListResponse> sourceListResponses = Lists.newArrayList(
- AutoPushSourceListResponse.builder()
+ ArrayList<StreamSource> streamSources = Lists.newArrayList(
+ AutoPushSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -304,7 +300,7 @@ class InnerInlongManagerClientTest {
.version(1)
.build(),
- MySQLBinlogSourceListResponse.builder()
+ MySQLBinlogSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -317,7 +313,7 @@ class InnerInlongManagerClientTest {
.tableWhiteList("")
.build(),
- FileSourceListResponse.builder()
+ FileSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -327,7 +323,7 @@ class InnerInlongManagerClientTest {
.timeOffset("timeOffset")
.build(),
- KafkaSourceListResponse.builder()
+ KafkaSource.builder()
.id(22)
.inlongGroupId("1")
.inlongStreamId("2")
@@ -335,7 +331,6 @@ class InnerInlongManagerClientTest {
.sourceName("source name")
.serializationType("csv")
.dataNodeName("dataNodeName")
-
.topic("topic")
.groupId("111")
.bootstrapServers("bootstrapServers")
@@ -349,7 +344,7 @@ class InnerInlongManagerClientTest {
.inlongGroupId("1")
.name("name")
.inCharges("admin")
- .sourceResponses(sourceListResponses)
+ .streamSources(streamSources)
.build()
);
@@ -390,7 +385,7 @@ class InnerInlongManagerClientTest {
)
);
- String groupId = innerInlongManagerClient.createGroup(new InlongGroupRequest());
+ String groupId = innerInlongManagerClient.createGroup(new InlongPulsarRequest());
Assertions.assertEquals("1111", groupId);
}
@@ -403,7 +398,7 @@ class InnerInlongManagerClientTest {
)
);
- Pair<String, String> updateGroup = innerInlongManagerClient.updateGroup(new InlongGroupRequest());
+ Pair<String, String> updateGroup = innerInlongManagerClient.updateGroup(new InlongPulsarRequest());
Assertions.assertEquals("1111", updateGroup.getKey());
Assertions.assertTrue(StringUtils.isBlank(updateGroup.getValue()));
}
@@ -676,9 +671,8 @@ class InnerInlongManagerClientTest {
)
)
);
- ClusterRequest request = new ClusterRequest();
+ ClusterRequest request = new PulsarClusterRequest();
request.setName("pulsar");
- request.setType("PULSAR");
request.setClusterTags("test_cluster");
Integer clusterIndex = innerInlongManagerClient.saveCluster(request);
Assertions.assertEquals(1, (int) clusterIndex);
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterInfo.java
index 8fc65460e..620c29135 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterInfo.java
@@ -24,7 +24,6 @@ import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import org.apache.inlong.manager.common.util.CommonBeanUtils;
import java.util.Date;
@@ -36,7 +35,7 @@ import java.util.Date;
@AllArgsConstructor
@ApiModel("Inlong cluster info")
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "type")
-public class ClusterInfo {
+public abstract class ClusterInfo {
@ApiModelProperty(value = "Primary key")
private Integer id;
@@ -83,8 +82,6 @@ public class ClusterInfo {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date modifyTime;
- public ClusterRequest genRequest() {
- return CommonBeanUtils.copyProperties(this, ClusterRequest::new);
- }
+ public abstract ClusterRequest genRequest();
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterRequest.java
index f860e6a8f..6e28579cf 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/cluster/ClusterRequest.java
@@ -21,7 +21,6 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.inlong.manager.common.pojo.common.UpdateValidation;
@@ -33,12 +32,11 @@ import javax.validation.constraints.NotNull;
* Inlong cluster request
*/
@Data
-@Builder
@NoArgsConstructor
@AllArgsConstructor
@ApiModel("Inlong cluster request")
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "type")
-public class ClusterRequest {
+public abstract class ClusterRequest {
@NotNull(groups = UpdateValidation.class)
@ApiModelProperty(value = "Primary key")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/consumption/ConsumptionMqExtBase.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/consumption/ConsumptionMqExtBase.java
index 6bbb13e00..f1206ca5e 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/consumption/ConsumptionMqExtBase.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/consumption/ConsumptionMqExtBase.java
@@ -19,7 +19,6 @@ package org.apache.inlong.manager.common.pojo.consumption;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
@@ -29,7 +28,8 @@ import lombok.Data;
*/
@Data
@ApiModel("Extended consumption information of different MQs")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "mqType", defaultImpl = ConsumptionMqExtBase.class)
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "mqType",
+ defaultImpl = ConsumptionMqExtBase.class)
@JsonSubTypes({
@JsonSubTypes.Type(value = ConsumptionPulsarInfo.class, name = "PULSAR"),
@JsonSubTypes.Type(value = ConsumptionPulsarInfo.class, name = "TDMQ_PULSAR")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
index 22caf3147..e2e9a6760 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
@@ -20,7 +20,6 @@ package org.apache.inlong.manager.common.pojo.group;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
@@ -41,7 +40,7 @@ import java.util.List;
@NoArgsConstructor
@AllArgsConstructor
@ApiModel("Inlong group info")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "mqType")
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "mqType")
public abstract class InlongGroupInfo {
@ApiModelProperty(value = "Primary key")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
index 8ed2b50d0..fb236d0b1 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
@@ -24,7 +24,7 @@ import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import java.util.Date;
import java.util.List;
@@ -61,6 +61,6 @@ public class InlongGroupListResponse {
private Date modifyTime;
@ApiModelProperty(value = "Stream sources in the inlong group")
- private List<SourceListResponse> sourceResponses;
+ private List<StreamSource> streamSources;
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
index a4e21b3ff..703590e4d 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
@@ -18,11 +18,9 @@
package org.apache.inlong.manager.common.pojo.group;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.hibernate.validator.constraints.Length;
@@ -35,12 +33,11 @@ import java.util.List;
* Inlong group request
*/
@Data
-@Builder
@NoArgsConstructor
@AllArgsConstructor
@ApiModel("Inlong group create request")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "mqType")
-public class InlongGroupRequest {
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "mqType")
+public abstract class InlongGroupRequest {
@NotBlank(message = "inlongGroupId cannot be blank")
@ApiModelProperty(value = "Inlong group id", required = true)
@@ -71,15 +68,12 @@ public class InlongGroupRequest {
private String tubeMaster;
@ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
- @Builder.Default
private Integer enableZookeeper = 0;
@ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
- @Builder.Default
private Integer enableCreateResource = 1;
@ApiModelProperty(value = "Whether to use lightweight mode, 0: false, 1: true")
- @Builder.Default
private Integer lightweight = 0;
@ApiModelProperty(value = "Inlong cluster tag, which links to inlong_cluster table")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
index c8f86d1c9..51c0243c5 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
@@ -19,7 +19,6 @@ package org.apache.inlong.manager.common.pojo.sink;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -36,8 +35,8 @@ import java.util.Map;
@SuperBuilder
@NoArgsConstructor
@AllArgsConstructor
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sinkType")
-public class SinkListResponse {
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sinkType")
+public abstract class SinkListResponse {
@ApiModelProperty(value = "Primary key")
private Integer id;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
index 72cbf5326..7a3652493 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
@@ -18,7 +18,6 @@
package org.apache.inlong.manager.common.pojo.sink;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import com.google.common.collect.Maps;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@@ -37,8 +36,8 @@ import java.util.Map;
*/
@Data
@ApiModel("Request of sink")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sinkType")
-public class SinkRequest {
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sinkType")
+public abstract class SinkRequest {
@NotNull(groups = UpdateValidation.class)
@ApiModelProperty(value = "Primary key")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/StreamSink.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/StreamSink.java
index 4997a2836..48ad58a30 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/StreamSink.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/StreamSink.java
@@ -20,7 +20,6 @@ package org.apache.inlong.manager.common.pojo.sink;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.swagger.annotations.ApiModel;
@@ -49,7 +48,7 @@ import java.util.Map;
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@ApiModel("Stream sink info")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sinkType")
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sinkType")
public abstract class StreamSink extends StreamNode {
@ApiModelProperty("Sink id")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sort/BaseSortConf.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sort/BaseSortConf.java
index e107d47af..27da8824d 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sort/BaseSortConf.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sort/BaseSortConf.java
@@ -18,7 +18,6 @@
package org.apache.inlong.manager.common.pojo.sort;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import lombok.Data;
@@ -27,7 +26,7 @@ import lombok.Data;
*/
@Data
@ApiModel("Sort configuration for inlong group")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sortType")
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sortType")
public abstract class BaseSortConf {
public String sortType;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceListResponse.java
deleted file mode 100644
index 17b123432..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceListResponse.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source;
-
-import com.fasterxml.jackson.annotation.JsonFormat;
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-import java.util.Date;
-
-/**
- * Response info of the source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@NoArgsConstructor
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sourceType")
-public class SourceListResponse {
-
- @ApiModelProperty(value = "Primary key")
- private Integer id;
-
- @ApiModelProperty(value = "Inlong group id")
- private String inlongGroupId;
-
- @ApiModelProperty(value = "Inlong stream id")
- private String inlongStreamId;
-
- @ApiModelProperty("Source type, including: FILE, KAFKA, etc.")
- private String sourceType;
-
- @ApiModelProperty("Source name, unique in one stream.")
- private String sourceName;
-
- @ApiModelProperty("Data Serialization, support: csv, json, canal, avro, etc")
- private String serializationType = "none";
-
- @ApiModelProperty("Data node name")
- private String dataNodeName;
-
- @ApiModelProperty("Id of the cluster that collected this source")
- private Integer clusterId;
-
- @ApiModelProperty(value = "Status")
- private Integer status;
-
- @ApiModelProperty("Version")
- private Integer version;
-
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
- private Date createTime;
-
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
- private Date modifyTime;
-
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceRequest.java
index ef3c77833..cec0e3a7d 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/SourceRequest.java
@@ -18,7 +18,6 @@
package org.apache.inlong.manager.common.pojo.source;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
@@ -36,7 +35,7 @@ import java.util.List;
*/
@Data
@ApiModel("Request of source")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sourceType")
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sourceType")
public class SourceRequest {
@NotNull(groups = UpdateValidation.class)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/StreamSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/StreamSource.java
index 064546b06..5f82fa6ea 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/StreamSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/StreamSource.java
@@ -19,7 +19,6 @@ package org.apache.inlong.manager.common.pojo.source;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
@@ -39,7 +38,7 @@ import java.util.Date;
@NoArgsConstructor
@AllArgsConstructor
@EqualsAndHashCode(callSuper = true)
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "sourceType")
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "sourceType")
@ApiModel("Stream source info")
public abstract class StreamSource extends StreamNode {
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/autopush/AutoPushSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/autopush/AutoPushSourceListResponse.java
deleted file mode 100644
index 36ce51b8f..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/autopush/AutoPushSourceListResponse.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.autopush;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response info of auto push source list.
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of auto push source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_AUTO_PUSH)
-public class AutoPushSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty(value = "DataProxy group name, used when the user enables local configuration")
- private String dataProxyGroup;
-
- public AutoPushSourceListResponse() {
- this.setSourceType(SourceType.AUTO_PUSH.getType());
- }
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/file/FileSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/file/FileSourceListResponse.java
deleted file mode 100644
index 8f69ed2a7..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/file/FileSourceListResponse.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.file;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response info of File source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of file source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_FILE)
-public class FileSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Agent IP address")
- private String ip;
-
- @ApiModelProperty("Path regex pattern for file, such as /a/b/*.txt")
- private String pattern;
-
- @ApiModelProperty("TimeOffset for collection, "
- + "'1m' means from one minute after, '-1m' means from one minute before, "
- + "'1h' means from one hour after, '-1h' means from one minute before, "
- + "'1d' means from one day after, '-1d' means from one minute before, "
- + "Null or blank means from current timestamp")
- private String timeOffset;
-
- public FileSourceListResponse() {
- this.setSourceType(SourceType.FILE.getType());
- }
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSource.java
index fecdc2bcb..e2cad1e94 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSource.java
@@ -57,10 +57,12 @@ public class KafkaSource extends StreamSource {
@ApiModelProperty("Limit the number of bytes read per second")
private String byteSpeedLimit;
- @ApiModelProperty("Topic partition offset")
+ @ApiModelProperty(value = "Topic partition offset",
+ notes = "For example, '0#100_1#10' means the offset of partition 0 is 100, the offset of partition 1 is 10")
private String topicPartitionOffset;
- @ApiModelProperty(value = "The strategy of auto offset reset")
+ @ApiModelProperty(value = "The strategy of auto offset reset",
+ notes = "including earliest, latest (the default), none")
private String autoOffsetReset;
@ApiModelProperty("database pattern used for filter in canal format")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSourceListResponse.java
deleted file mode 100644
index 89df235d6..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/kafka/KafkaSourceListResponse.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.kafka;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response of kafka source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of kafka source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_KAFKA)
-public class KafkaSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Kafka topic")
- private String topic;
-
- @ApiModelProperty("Kafka consumer group")
- private String groupId;
-
- @ApiModelProperty("Kafka servers address")
- private String bootstrapServers;
-
- @ApiModelProperty("Limit the amount of data read per second")
- private String recordSpeedLimit;
-
- @ApiModelProperty("Limit the number of bytes read per second")
- private String byteSpeedLimit;
-
- @ApiModelProperty(value = "Topic partition offset",
- notes = "For example, '0#100_1#10' means the offset of partition 0 is 100, the offset of partition 1 is 10")
- private String topicPartitionOffset;
-
- @ApiModelProperty(value = "The strategy of auto offset reset",
- notes = "including earliest, latest (the default), none")
- private String autoOffsetReset;
-
- @ApiModelProperty("Primary key, needed when serialization type is csv, json, avro")
- private String primaryKey;
-
- public KafkaSourceListResponse() {
- this.setSourceType(SourceType.KAFKA.getType());
- }
-
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSource.java
index 45c4b790c..cd23b6049 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSource.java
@@ -42,24 +42,24 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = SourceType.SOURCE_MONGODB)
public class MongoDBSource extends StreamSource {
- @ApiModelProperty("MongoDB primaryKey")
- private String primaryKey;
-
- @ApiModelProperty("MongoDB hosts")
+ @ApiModelProperty("Hosts of the MongoDB server")
private String hosts;
- @ApiModelProperty("MongoDB username")
+ @ApiModelProperty("Username of the MongoDB server")
private String username;
- @ApiModelProperty("MongoDB password")
+ @ApiModelProperty("Password of the MongoDB server")
private String password;
- @ApiModelProperty("MongoDB database")
+ @ApiModelProperty("MongoDB database name")
private String database;
- @ApiModelProperty("MongoDB collection")
+ @ApiModelProperty("MongoDB collection name")
private String collection;
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
public MongoDBSource() {
this.setSourceType(SourceType.MONGODB.name());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceDTO.java
index 20033831e..1e629755e 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceDTO.java
@@ -40,24 +40,24 @@ public class MongoDBSourceDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
- @ApiModelProperty("MongoDB primaryKey")
- private String primaryKey;
-
- @ApiModelProperty("MongoDB hosts")
+ @ApiModelProperty("Hosts of the MongoDB server")
private String hosts;
- @ApiModelProperty("MongoDB username")
+ @ApiModelProperty("Username of the MongoDB server")
private String username;
- @ApiModelProperty("MongoDB password")
+ @ApiModelProperty("Password of the MongoDB server")
private String password;
- @ApiModelProperty("MongoDB database")
+ @ApiModelProperty("MongoDB database name")
private String database;
- @ApiModelProperty("MongoDB collection")
+ @ApiModelProperty("MongoDB collection name")
private String collection;
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
/**
* Get the dto instance from the request
*/
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceListResponse.java
deleted file mode 100644
index c3451e761..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceListResponse.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.mongodb;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response of MongoDB paging list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of MongoDB paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_MONGODB)
-public class MongoDBSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("MongoDB primaryKey")
- private String primaryKey;
-
- @ApiModelProperty("MongoDB hosts")
- private String hosts;
-
- @ApiModelProperty("MongoDB username")
- private String username;
-
- @ApiModelProperty("MongoDB password")
- private String password;
-
- @ApiModelProperty("MongoDB database")
- private String database;
-
- @ApiModelProperty("MongoDB collection")
- private String collection;
-
- public MongoDBSourceListResponse() {
- this.setSourceType(SourceType.MONGODB.getType());
- }
-
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceRequest.java
index 83cd0bf77..b3dbbc933 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mongodb/MongoDBSourceRequest.java
@@ -36,24 +36,24 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = SourceType.SOURCE_MONGODB)
public class MongoDBSourceRequest extends SourceRequest {
- @ApiModelProperty("MongoDB primaryKey")
- private String primaryKey;
-
- @ApiModelProperty("MongoDB hosts")
+ @ApiModelProperty("Hosts of the MongoDB server")
private String hosts;
- @ApiModelProperty("MongoDB username")
+ @ApiModelProperty("Username of the MongoDB server")
private String username;
- @ApiModelProperty("MongoDB password")
+ @ApiModelProperty("Password of the MongoDB server")
private String password;
- @ApiModelProperty("MongoDB database")
+ @ApiModelProperty("MongoDB database name")
private String database;
- @ApiModelProperty("MongoDB collection")
+ @ApiModelProperty("MongoDB collection name")
private String collection;
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
public MongoDBSourceRequest() {
this.setSourceType(SourceType.MONGODB.toString());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSource.java
index daa009403..c10b0719c 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSource.java
@@ -42,17 +42,17 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = SourceType.SOURCE_BINLOG)
public class MySQLBinlogSource extends StreamSource {
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the MySQL server")
private String user;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the MySQL server")
private String password;
- @ApiModelProperty("Hostname of the DB server")
+ @ApiModelProperty("Hostname of the MySQL server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
+ @ApiModelProperty("Port of the MySQL server")
+ private Integer port;
@ApiModelProperty("Id of physical node of MySQL Cluster, 0 if single node")
private Integer serverId = 0;
@@ -90,14 +90,14 @@ public class MySQLBinlogSource extends StreamSource {
@ApiModelProperty("Need transfer total database")
private boolean allMigration;
- @ApiModelProperty(value = "Primary key must be shared by all tables", required = false)
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
@ApiModelProperty("Directly read binlog from the specified offset filename")
- public String specificOffsetFile;
+ private String specificOffsetFile;
@ApiModelProperty("Directly read binlog from the specified offset position")
- public Integer specificOffsetPos;
+ private Integer specificOffsetPos;
public MySQLBinlogSource() {
this.setSourceType(SourceType.BINLOG.name());
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceDTO.java
index 9b17880bd..23bc07a07 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceDTO.java
@@ -40,17 +40,17 @@ public class MySQLBinlogSourceDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); // thread safe
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the MySQL server")
private String user;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the MySQL server")
private String password;
- @ApiModelProperty("Hostname of the DB server")
+ @ApiModelProperty("Hostname of the MySQL server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
+ @ApiModelProperty("Port of the MySQL server")
+ private Integer port;
@ApiModelProperty("Id of physical node of MySQL Cluster, 0 if single node")
@Builder.Default
@@ -109,14 +109,14 @@ public class MySQLBinlogSourceDTO {
@ApiModelProperty("Whether to migrate all databases")
private boolean allMigration;
- @ApiModelProperty(value = "Primary key must be shared by all tables", required = false)
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
@ApiModelProperty("Directly read binlog from the specified offset filename")
- public String specificOffsetFile;
+ private String specificOffsetFile;
@ApiModelProperty("Directly read binlog from the specified offset position")
- public Integer specificOffsetPos;
+ private Integer specificOffsetPos;
/**
* Get the dto instance from the request
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceListResponse.java
deleted file mode 100644
index c840185a1..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceListResponse.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.mysql;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response info of binlog source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of binlog source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_BINLOG)
-public class MySQLBinlogSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Username of the DB server")
- private String user;
-
- @ApiModelProperty("Password of the DB server")
- private String password;
-
- @ApiModelProperty("Hostname of the DB server")
- private String hostname;
-
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
-
- @ApiModelProperty("Id of physical node of MySQL Cluster, 0 if single node")
- private Integer serverId = 0;
-
- @ApiModelProperty("Whether include schema, default is 'false'")
- private String includeSchema;
-
- @ApiModelProperty(value = "List of DBs to be collected, seperated by ',', supporting regular expressions")
- private String databaseWhiteList;
-
- @ApiModelProperty(value = "List of tables to be collected, seperated by ',',supporting regular expressions")
- private String tableWhiteList;
-
- @ApiModelProperty("Database time zone, Default is UTC")
- private String serverTimezone;
-
- @ApiModelProperty("The interval for recording an offset")
- private String intervalMs;
-
- @ApiModelProperty("Snapshot mode, supports: initial, when_needed, never, schema_only, schema_only_recovery")
- private String snapshotMode;
-
- @ApiModelProperty("The file path to store offset info")
- private String offsetFilename;
-
- @ApiModelProperty("The file path to store history info")
- private String historyFilename;
-
- @ApiModelProperty("Whether to monitor the DDL, default is 'false'")
- private String monitoredDdl;
-
- @ApiModelProperty("Timestamp standard for binlog: SQL, ISO_8601")
- private String timestampFormatStandard = "SQL";
-
- @ApiModelProperty("Need transfer total database")
- private boolean allMigration;
-
- @ApiModelProperty(value = "Primary key must be shared by all tables", required = false)
- private String primaryKey;
-
- @ApiModelProperty("Directly read binlog from the specified offset filename")
- public String specificOffsetFile;
-
- @ApiModelProperty("Directly read binlog from the specified offset position")
- public Integer specificOffsetPos;
-
- public MySQLBinlogSourceListResponse() {
- this.setSourceType(SourceType.BINLOG.getType());
- }
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceRequest.java
index 636f7b56c..1ef2bce14 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/mysql/MySQLBinlogSourceRequest.java
@@ -46,8 +46,8 @@ public class MySQLBinlogSourceRequest extends SourceRequest {
@ApiModelProperty("Hostname of the DB server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port = 3306;
+ @ApiModelProperty("Port of the DB server")
+ private Integer port = 3306;
@ApiModelProperty("Id of physical node of MySQL Cluster, 0 if single node")
private Integer serverId = 0;
@@ -108,10 +108,10 @@ public class MySQLBinlogSourceRequest extends SourceRequest {
private String primaryKey;
@ApiModelProperty("Directly read binlog from the specified offset filename")
- public String specificOffsetFile;
+ private String specificOffsetFile;
@ApiModelProperty("Directly read binlog from the specified offset position")
- public Integer specificOffsetPos;
+ private Integer specificOffsetPos;
public MySQLBinlogSourceRequest() {
this.setSourceType(SourceType.BINLOG.toString());
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSource.java
index 8c7cc800d..0815b868b 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSource.java
@@ -42,16 +42,16 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = SourceType.SOURCE_ORACLE)
public class OracleSource extends StreamSource {
- @ApiModelProperty("Hostname of the DB server, for example: 127.0.0.1")
+ @ApiModelProperty("Hostname of the Oracle server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private Integer port = 1521;
+ @ApiModelProperty("Port of the Oracle server")
+ private Integer port;
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the Oracle server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the Oracle server")
private String password;
@ApiModelProperty("Database name")
@@ -60,13 +60,13 @@ public class OracleSource extends StreamSource {
@ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("table name")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Scan startup mode")
private String scanStartupMode;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
@ApiModelProperty("Need transfer total database")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceDTO.java
index 2b125b7c9..02c7e29bf 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceDTO.java
@@ -40,16 +40,16 @@ public class OracleSourceDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
- @ApiModelProperty("Hostname of the DB server, for example: 127.0.0.1")
+ @ApiModelProperty("Hostname of the Oracle server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private Integer port = 1521;
+ @ApiModelProperty("Port of the Oracle server")
+ private Integer port;
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the Oracle server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the Oracle server")
private String password;
@ApiModelProperty("Database name")
@@ -58,13 +58,13 @@ public class OracleSourceDTO {
@ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("table name")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Scan startup mode")
private String scanStartupMode;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
/**
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceListResponse.java
deleted file mode 100644
index d89a9632f..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceListResponse.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.oracle;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response info of oracle source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of oracle source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_ORACLE)
-public class OracleSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Hostname of the DB server, for example: 127.0.0.1")
- private String hostname;
-
- @ApiModelProperty("Exposed port of the DB server")
- private Integer port = 1521;
-
- @ApiModelProperty("Username of the DB server")
- private String username;
-
- @ApiModelProperty("Password of the DB server")
- private String password;
-
- @ApiModelProperty("Database name")
- private String database;
-
- @ApiModelProperty("Schema name")
- private String schemaName;
-
- @ApiModelProperty("table name")
- private String tableName;
-
- @ApiModelProperty("Scan startup mode")
- private String scanStartupMode;
-
- @ApiModelProperty(value = "Primary key must be shared by all tables")
- private String primaryKey;
-
- @ApiModelProperty("Need transfer total database")
- private boolean allMigration = false;
-
- public OracleSourceListResponse() {
- this.setSourceType(SourceType.ORACLE.getType());
- }
-
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceRequest.java
index e37e0b348..5a7db9ed1 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/oracle/OracleSourceRequest.java
@@ -27,7 +27,7 @@ import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.util.JsonTypeDefine;
/**
- * Request of oracle source
+ * Request of Oracle source
*/
@Data
@ToString(callSuper = true)
@@ -36,16 +36,16 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = SourceType.SOURCE_ORACLE)
public class OracleSourceRequest extends SourceRequest {
- @ApiModelProperty("Hostname of the DB server, for example: 127.0.0.1")
+ @ApiModelProperty("Hostname of the Oracle server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
+ @ApiModelProperty("Port of the Oracle server")
private Integer port = 1521;
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the Oracle server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the Oracle server")
private String password;
@ApiModelProperty("Database name")
@@ -54,13 +54,13 @@ public class OracleSourceRequest extends SourceRequest {
@ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("table name")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Scan startup mode")
private String scanStartupMode;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
@ApiModelProperty("Need transfer total database")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceListResponse.java
deleted file mode 100644
index bf493504e..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceListResponse.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.postgres;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-import java.util.List;
-
-/**
- * Response info of postgres source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of postgres source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_POSTGRES)
-public class PostgresSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Primary key")
- private String primaryKey;
-
- @ApiModelProperty("Username of the DB server")
- private String username;
-
- @ApiModelProperty("Password of the DB server")
- private String password;
-
- @ApiModelProperty("Hostname of the DB server")
- private String hostname;
-
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
-
- @ApiModelProperty("Exposed database of the DB")
- private String database;
-
- @ApiModelProperty("Schema info")
- private String schema;
-
- @ApiModelProperty("Decoding plugin name")
- private String decodingPluginName;
-
- @ApiModelProperty("List of tables name")
- private List<String> tableNameList;
-
- public PostgresSourceListResponse() {
- this.setSourceType(SourceType.POSTGRES.getType());
- }
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSource.java
similarity index 72%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSource.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSource.java
index ac4efdd50..4c3a1c426 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSource.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.common.pojo.source.postgres;
+package org.apache.inlong.manager.common.pojo.source.postgresql;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@@ -33,51 +33,51 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
import java.util.List;
/**
- * Postgres source info
+ * PostgreSQL source info
*/
@Data
@SuperBuilder
@AllArgsConstructor
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
-@ApiModel(value = "Postgres source info")
+@ApiModel(value = "PostgreSQL source info")
@JsonTypeDefine(value = SourceType.SOURCE_POSTGRES)
-public class PostgresSource extends StreamSource {
+public class PostgreSQLSource extends StreamSource {
- @ApiModelProperty("Primary key")
- private String primaryKey;
-
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the PostgreSQL server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the PostgreSQL server")
private String password;
- @ApiModelProperty("Hostname of the DB server")
+ @ApiModelProperty("Hostname of the PostgreSQL server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
+ @ApiModelProperty("Port of the PostgreSQL server")
+ private Integer port = 5432;
- @ApiModelProperty("Exposed database of the DB")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema info")
+ @ApiModelProperty("Schema name")
private String schema;
@ApiModelProperty("Decoding plugin name")
private String decodingPluginName;
- @ApiModelProperty("List of tables name")
+ @ApiModelProperty("List of table name")
private List<String> tableNameList;
- public PostgresSource() {
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
+ public PostgreSQLSource() {
this.setSourceType(SourceType.POSTGRES.name());
}
@Override
public SourceRequest genSourceRequest() {
- return CommonBeanUtils.copyProperties(this, PostgresSourceRequest::new);
+ return CommonBeanUtils.copyProperties(this, PostgreSQLSourceRequest::new);
}
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceDTO.java
similarity index 74%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceDTO.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceDTO.java
index efe184603..69b70e9b3 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceDTO.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.common.pojo.source.postgres;
+package org.apache.inlong.manager.common.pojo.source.postgresql;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -31,48 +31,48 @@ import javax.validation.constraints.NotNull;
import java.util.List;
/**
- * Postgres source info
+ * PostgreSQL source info
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class PostgresSourceDTO {
+public class PostgreSQLSourceDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); // thread safe
- @ApiModelProperty("Primary key")
- private String primaryKey;
-
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the PostgreSQL server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the PostgreSQL server")
private String password;
- @ApiModelProperty("Hostname of the DB server")
+ @ApiModelProperty("Hostname of the PostgreSQL server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
+ @ApiModelProperty("Port of the PostgreSQL server")
+ private Integer port;
- @ApiModelProperty("Exposed database of the DB")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema info")
+ @ApiModelProperty("Schema name")
private String schema;
@ApiModelProperty("Decoding plugin name")
private String decodingPluginName;
- @ApiModelProperty("List of tables name")
+ @ApiModelProperty("List of table name")
private List<String> tableNameList;
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
/**
* Get the dto instance from the request
*/
- public static PostgresSourceDTO getFromRequest(PostgresSourceRequest request) {
- return PostgresSourceDTO.builder()
+ public static PostgreSQLSourceDTO getFromRequest(PostgreSQLSourceRequest request) {
+ return PostgreSQLSourceDTO.builder()
.username(request.getUsername())
.password(request.getPassword())
.hostname(request.getHostname())
@@ -85,10 +85,10 @@ public class PostgresSourceDTO {
.build();
}
- public static PostgresSourceDTO getFromJson(@NotNull String extParams) {
+ public static PostgreSQLSourceDTO getFromJson(@NotNull String extParams) {
try {
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
- return OBJECT_MAPPER.readValue(extParams, PostgresSourceDTO.class);
+ return OBJECT_MAPPER.readValue(extParams, PostgreSQLSourceDTO.class);
} catch (Exception e) {
throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceRequest.java
similarity index 70%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceRequest.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceRequest.java
index f12ea5640..4d2976d38 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgres/PostgresSourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/postgresql/PostgreSQLSourceRequest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.common.pojo.source.postgres;
+package org.apache.inlong.manager.common.pojo.source.postgresql;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@@ -29,43 +29,43 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
import java.util.List;
/**
- * Request of postgres source
+ * Request of PostgreSQL source
*/
@Data
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
-@ApiModel(value = "Request of the postgres source info")
+@ApiModel(value = "Request of the PostgreSQL source")
@JsonTypeDefine(value = SourceType.SOURCE_POSTGRES)
-public class PostgresSourceRequest extends SourceRequest {
+public class PostgreSQLSourceRequest extends SourceRequest {
- @ApiModelProperty("Primary key")
- private String primaryKey;
-
- @ApiModelProperty("Username of the DB server")
+ @ApiModelProperty("Username of the PostgreSQL server")
private String username;
- @ApiModelProperty("Password of the DB server")
+ @ApiModelProperty("Password of the PostgreSQL server")
private String password;
- @ApiModelProperty("Hostname of the DB server")
+ @ApiModelProperty("Hostname of the PostgreSQL server")
private String hostname;
- @ApiModelProperty("Exposed port of the DB server")
- private int port;
+ @ApiModelProperty("Port of the PostgreSQL server")
+ private Integer port = 5432;
- @ApiModelProperty("Exposed database of the DB")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema info")
+ @ApiModelProperty("Schema name")
private String schema;
@ApiModelProperty("Decoding plugin name")
private String decodingPluginName;
- @ApiModelProperty("List of tables name")
+ @ApiModelProperty("List of table name")
private List<String> tableNameList;
- public PostgresSourceRequest() {
+ @ApiModelProperty("Primary key must be shared by all tables")
+ private String primaryKey;
+
+ public PostgreSQLSourceRequest() {
this.setSourceType(SourceType.POSTGRES.toString());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/pulsar/PulsarSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/pulsar/PulsarSourceListResponse.java
deleted file mode 100644
index 6143f23dd..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/pulsar/PulsarSourceListResponse.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.pulsar;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response of pulsar source list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of pulsar source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_PULSAR)
-public class PulsarSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Pulsar tenant")
- private String tenant = "default";
-
- @ApiModelProperty("Pulsar namespace")
- private String namespace;
-
- @ApiModelProperty("Pulsar topic")
- private String topic;
-
- @ApiModelProperty("Pulsar adminUrl")
- private String adminUrl;
-
- @ApiModelProperty("Pulsar serviceUrl")
- private String serviceUrl;
-
- @ApiModelProperty("Primary key, needed when serialization type is csv, json, avro")
- private String primaryKey;
-
- @ApiModelProperty("Configure the Source's startup mode."
- + " Available options are earliest, latest, external-subscription, and specific-offsets.")
- private String scanStartupMode = "earliest";
-
- public PulsarSourceListResponse() {
- this.setSourceType(SourceType.PULSAR.getType());
- }
-}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSource.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSource.java
similarity index 75%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSource.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSource.java
index 86fafc5d6..14f35dabb 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSource.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSource.java
@@ -31,36 +31,36 @@ import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.JsonTypeDefine;
/**
- * SqlServer source info
+ * SQLServer source info
*/
@Data
@SuperBuilder
@AllArgsConstructor
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
-@ApiModel(value = "SqlServer source info")
+@ApiModel(value = "SQLServer source info")
@JsonTypeDefine(value = SourceType.SOURCE_SQLSERVER)
-public class SqlServerSource extends StreamSource {
+public class SQLServerSource extends StreamSource {
- @ApiModelProperty("Username of the SqlServer")
+ @ApiModelProperty("Username of the SQLServer server")
private String username;
- @ApiModelProperty("Password of the SqlServer")
+ @ApiModelProperty("Password of the SQLServer server")
private String password;
- @ApiModelProperty("Hostname of the SqlServer")
+ @ApiModelProperty("Hostname of the SQLServer server")
private String hostname;
- @ApiModelProperty("Exposed port of the SqlServer")
- private int port;
+ @ApiModelProperty("Port of the SQLServer server")
+ private Integer port;
- @ApiModelProperty("Database of the SqlServer")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema name of the SqlServer")
+ @ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("Table name of the SqlServer")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Database time zone, default is UTC")
@@ -69,16 +69,16 @@ public class SqlServerSource extends StreamSource {
@ApiModelProperty("Whether to migrate all databases")
private boolean allMigration;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
- public SqlServerSource() {
+ public SQLServerSource() {
this.setSourceType(SourceType.SQLSERVER.name());
}
@Override
public SourceRequest genSourceRequest() {
- return CommonBeanUtils.copyProperties(this, SqlServerSourceRequest::new);
+ return CommonBeanUtils.copyProperties(this, SQLServerSourceRequest::new);
}
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceDTO.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceDTO.java
similarity index 78%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceDTO.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceDTO.java
index 653fe5543..03b6177a8 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceDTO.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceDTO.java
@@ -30,35 +30,35 @@ import org.apache.inlong.manager.common.exceptions.BusinessException;
import javax.validation.constraints.NotNull;
/**
- * Sqlserver source info
+ * SQLServer source info
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class SqlServerSourceDTO {
+public class SQLServerSourceDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
- @ApiModelProperty("Username of the SqlServer")
+ @ApiModelProperty("Username of the SQLServer server")
private String username;
- @ApiModelProperty("Password of the SqlServer")
+ @ApiModelProperty("Password of the SQLServer server")
private String password;
- @ApiModelProperty("Hostname of the SqlServer")
+ @ApiModelProperty("Hostname of the SQLServer server")
private String hostname;
- @ApiModelProperty("Exposed port of the SqlServer")
- private int port;
+ @ApiModelProperty("Port of the SQLServer server")
+ private Integer port;
- @ApiModelProperty("Database of the SqlServer")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema name of the SqlServer")
+ @ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("Table name of the SqlServer")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Database time zone, default is UTC")
@@ -67,14 +67,14 @@ public class SqlServerSourceDTO {
@ApiModelProperty("Whether to migrate all databases")
private boolean allMigration;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
/**
* Get the dto instance from the request
*/
- public static SqlServerSourceDTO getFromRequest(SqlServerSourceRequest request) {
- return SqlServerSourceDTO.builder()
+ public static SQLServerSourceDTO getFromRequest(SQLServerSourceRequest request) {
+ return SQLServerSourceDTO.builder()
.username(request.getUsername())
.password(request.getPassword())
.hostname(request.getHostname())
@@ -91,10 +91,10 @@ public class SqlServerSourceDTO {
/**
* Get the dto instance from the JSON string
*/
- public static SqlServerSourceDTO getFromJson(@NotNull String extParams) {
+ public static SQLServerSourceDTO getFromJson(@NotNull String extParams) {
try {
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
- return OBJECT_MAPPER.readValue(extParams, SqlServerSourceDTO.class);
+ return OBJECT_MAPPER.readValue(extParams, SQLServerSourceDTO.class);
} catch (Exception e) {
throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceRequest.java
similarity index 77%
rename from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceRequest.java
rename to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceRequest.java
index 351be9386..1800c410f 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SQLServerSourceRequest.java
@@ -34,27 +34,27 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@EqualsAndHashCode(callSuper = true)
@ApiModel(value = "Request of the SqlServer source")
@JsonTypeDefine(value = SourceType.SOURCE_SQLSERVER)
-public class SqlServerSourceRequest extends SourceRequest {
+public class SQLServerSourceRequest extends SourceRequest {
- @ApiModelProperty("Username of the SqlServer")
+ @ApiModelProperty("Username of the SQLServer server")
private String username;
- @ApiModelProperty("Password of the SqlServer")
+ @ApiModelProperty("Password of the SQLServer server")
private String password;
- @ApiModelProperty("Hostname of the SqlServer")
+ @ApiModelProperty("Hostname of the SQLServer server")
private String hostname;
- @ApiModelProperty("Exposed port of the SqlServer")
- private int port;
+ @ApiModelProperty("Port of the SQLServer server")
+ private Integer port = 1433;
- @ApiModelProperty("Database of the SqlServer")
+ @ApiModelProperty("Database name")
private String database;
- @ApiModelProperty("Schema name of the SqlServer")
+ @ApiModelProperty("Schema name")
private String schemaName;
- @ApiModelProperty("Table name of the SqlServer")
+ @ApiModelProperty("Table name")
private String tableName;
@ApiModelProperty("Database time zone, default is UTC")
@@ -63,10 +63,10 @@ public class SqlServerSourceRequest extends SourceRequest {
@ApiModelProperty("Whether to migrate all databases")
private boolean allMigration;
- @ApiModelProperty(value = "Primary key must be shared by all tables")
+ @ApiModelProperty("Primary key must be shared by all tables")
private String primaryKey;
- public SqlServerSourceRequest() {
+ public SQLServerSourceRequest() {
this.setSourceType(SourceType.SQLSERVER.toString());
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceListResponse.java
deleted file mode 100644
index 583b0d576..000000000
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/sqlserver/SqlServerSourceListResponse.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.common.pojo.source.sqlserver;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.experimental.SuperBuilder;
-import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
-import org.apache.inlong.manager.common.util.JsonTypeDefine;
-
-/**
- * Response of SqlServer source paging list
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Response of SqlServer source paging list")
-@JsonTypeDefine(value = SourceType.SOURCE_SQLSERVER)
-public class SqlServerSourceListResponse extends SourceListResponse {
-
- @ApiModelProperty("Username of the SqlServer")
- private String username;
-
- @ApiModelProperty("Password of the SqlServer")
- private String password;
-
- @ApiModelProperty("Hostname of the SqlServer")
- private String hostname;
-
- @ApiModelProperty("Exposed port of the SqlServer")
- private int port;
-
- @ApiModelProperty("Database of the SqlServer")
- private String database;
-
- @ApiModelProperty("Schema name of the SqlServer")
- private String schemaName;
-
- @ApiModelProperty("Table name of the SqlServer")
- private String tableName;
-
- @ApiModelProperty("Database time zone, default is UTC")
- private String serverTimezone;
-
- @ApiModelProperty("Whether to migrate all databases")
- private boolean allMigration;
-
- @ApiModelProperty(value = "Primary key must be shared by all tables")
- private String primaryKey;
-
- public SqlServerSourceListResponse() {
- this.setSourceType(SourceType.SQLSERVER.getType());
- }
-
-}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
index 26bc00dfa..f90573811 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
@@ -36,7 +36,7 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
import org.apache.inlong.manager.common.pojo.group.InlongGroupRequest;
import org.apache.inlong.manager.common.pojo.group.InlongGroupTopicInfo;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
@@ -183,18 +183,16 @@ public class InlongGroupServiceImpl implements InlongGroupService {
Set<String> groupIds = groupResponseList.stream().map(InlongGroupListResponse::getInlongGroupId)
.collect(Collectors.toSet());
List<StreamSourceEntity> sourceEntities = streamSourceMapper.selectByGroupIds(new ArrayList<>(groupIds));
- Map<String, List<SourceListResponse>> sourceMap = Maps.newHashMap();
+ Map<String, List<StreamSource>> sourceMap = Maps.newHashMap();
sourceEntities.forEach(sourceEntity -> {
SourceType sourceType = SourceType.forType(sourceEntity.getSourceType());
StreamSourceOperator operation = sourceOperatorFactory.getInstance(sourceType);
- SourceListResponse sourceListResponse = operation.getFromEntity(sourceEntity, SourceListResponse::new);
- sourceMap.computeIfAbsent(sourceEntity.getInlongGroupId(), k -> Lists.newArrayList())
- .add(sourceListResponse);
+ StreamSource source = operation.getFromEntity(sourceEntity);
+ sourceMap.computeIfAbsent(sourceEntity.getInlongGroupId(), k -> Lists.newArrayList()).add(source);
});
groupResponseList.forEach(group -> {
- List<SourceListResponse> sourceListResponses = sourceMap.getOrDefault(group.getInlongGroupId(),
- Lists.newArrayList());
- group.setSourceResponses(sourceListResponses);
+ List<StreamSource> sources = sourceMap.getOrDefault(group.getInlongGroupId(), Lists.newArrayList());
+ group.setStreamSources(sources);
});
}
PageInfo<InlongGroupListResponse> page = new PageInfo<>(groupResponseList);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
index 08988fa3d..8c4e6c756 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
@@ -31,9 +31,9 @@ import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSource;
import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSource;
import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSource;
import org.apache.inlong.manager.common.pojo.source.oracle.OracleSource;
-import org.apache.inlong.manager.common.pojo.source.postgres.PostgresSource;
+import org.apache.inlong.manager.common.pojo.source.postgresql.PostgreSQLSource;
import org.apache.inlong.manager.common.pojo.source.pulsar.PulsarSource;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSource;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSource;
import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.constant.OracleConstant.ScanStartUpMode;
@@ -86,11 +86,11 @@ public class ExtractNodeUtils {
case PULSAR:
return createExtractNode((PulsarSource) sourceInfo);
case POSTGRES:
- return createExtractNode((PostgresSource) sourceInfo);
+ return createExtractNode((PostgreSQLSource) sourceInfo);
case ORACLE:
return createExtractNode((OracleSource) sourceInfo);
case SQLSERVER:
- return createExtractNode((SqlServerSource) sourceInfo);
+ return createExtractNode((SQLServerSource) sourceInfo);
case MONGODB:
return createExtractNode((MongoDBSource) sourceInfo);
default:
@@ -281,22 +281,22 @@ public class ExtractNodeUtils {
/**
* Create PostgreSQL extract node
*
- * @param postgresSource PostgreSQL source info
+ * @param postgreSQLSource PostgreSQL source info
* @return PostgreSQL extract node info
*/
- public static PostgresExtractNode createExtractNode(PostgresSource postgresSource) {
- List<StreamField> streamFields = postgresSource.getFieldList();
- String id = postgresSource.getSourceName();
- String name = postgresSource.getSourceName();
+ public static PostgresExtractNode createExtractNode(PostgreSQLSource postgreSQLSource) {
+ List<StreamField> streamFields = postgreSQLSource.getFieldList();
+ String id = postgreSQLSource.getSourceName();
+ String name = postgreSQLSource.getSourceName();
List<FieldInfo> fields = streamFields.stream()
.map(streamFieldInfo -> FieldInfoUtils.parseStreamFieldInfo(streamFieldInfo, name))
.collect(Collectors.toList());
return new PostgresExtractNode(id, name, fields, null, null,
- postgresSource.getPrimaryKey(), postgresSource.getTableNameList(),
- postgresSource.getHostname(), postgresSource.getUsername(),
- postgresSource.getPassword(), postgresSource.getDatabase(),
- postgresSource.getSchema(), postgresSource.getPort(),
- postgresSource.getDecodingPluginName());
+ postgreSQLSource.getPrimaryKey(), postgreSQLSource.getTableNameList(),
+ postgreSQLSource.getHostname(), postgreSQLSource.getUsername(),
+ postgreSQLSource.getPassword(), postgreSQLSource.getDatabase(),
+ postgreSQLSource.getSchema(), postgreSQLSource.getPort(),
+ postgreSQLSource.getDecodingPluginName());
}
/**
@@ -340,7 +340,7 @@ public class ExtractNodeUtils {
* @param source SqlServer source info
* @return SqlServer extract node info
*/
- public static SqlServerExtractNode createExtractNode(SqlServerSource source) {
+ public static SqlServerExtractNode createExtractNode(SQLServerSource source) {
String name = source.getSourceName();
List<StreamField> streamFields = source.getFieldList();
List<FieldInfo> fieldInfos = streamFields.stream()
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
index 65f5779bb..ddb3a18af 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
@@ -17,6 +17,8 @@
package org.apache.inlong.manager.service.source;
+import com.github.pagehelper.Page;
+import com.github.pagehelper.PageInfo;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.manager.common.consts.InlongConstants;
@@ -37,10 +39,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Isolation;
-import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
-import javax.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
@@ -72,13 +72,6 @@ public abstract class AbstractSourceOperator implements StreamSourceOperator {
*/
protected abstract String getSourceType();
- /**
- * Creating source object.
- *
- * @return source object
- */
- protected abstract StreamSource getSource();
-
@Override
@Transactional(rollbackFor = Throwable.class)
public Integer saveOpt(SourceRequest request, Integer groupStatus, String operator) {
@@ -112,19 +105,17 @@ public abstract class AbstractSourceOperator implements StreamSourceOperator {
}
@Override
- @Transactional(rollbackFor = Throwable.class, propagation = Propagation.NOT_SUPPORTED)
- public StreamSource getByEntity(@NotNull StreamSourceEntity entity) {
- Preconditions.checkNotNull(entity, ErrorCodeEnum.SOURCE_INFO_NOT_FOUND.getMessage());
- String existType = entity.getSourceType();
- Preconditions.checkTrue(getSourceType().equals(existType),
- String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
+ public List<StreamField> getSourceFields(Integer sourceId) {
+ List<StreamSourceFieldEntity> sourceFieldEntities = sourceFieldMapper.selectBySourceId(sourceId);
+ return CommonBeanUtils.copyListProperties(sourceFieldEntities, StreamField::new);
+ }
- StreamSource source = this.getFromEntity(entity, this::getSource);
- List<StreamSourceFieldEntity> sourceFieldEntities = sourceFieldMapper.selectBySourceId(entity.getId());
- List<StreamField> fieldInfos = CommonBeanUtils.copyListProperties(sourceFieldEntities,
- StreamField::new);
- source.setFieldList(fieldInfos);
- return source;
+ @Override
+ public PageInfo<? extends StreamSource> getPageInfo(Page<StreamSourceEntity> entityPage) {
+ if (CollectionUtils.isEmpty(entityPage)) {
+ return new PageInfo<>();
+ }
+ return entityPage.toPageInfo(this::getFromEntity);
}
@Override
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceOperator.java
index 234eebd32..09aa848b1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceOperator.java
@@ -20,12 +20,13 @@ package org.apache.inlong.manager.service.source;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageInfo;
import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
-import java.util.function.Supplier;
+import javax.validation.constraints.NotNull;
+import java.util.List;
/**
* Interface of the source operator
@@ -53,17 +54,15 @@ public interface StreamSourceOperator {
* @param entity get field value from the entity
* @return source info
*/
- StreamSource getByEntity(StreamSourceEntity entity);
+ StreamSource getFromEntity(StreamSourceEntity entity);
/**
- * Get the target from the given entity.
+ * Get stream source field list by the given source id.
*
- * @param entity get field value from the entity
- * @param target encapsulate value to the target
- * @param <T> target type
- * @return target after encapsulating.
+ * @param sourceId source id
+ * @return stream field list
*/
- <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target);
+ List<StreamField> getSourceFields(@NotNull Integer sourceId);
/**
* Get source list response from the given source entity page.
@@ -71,7 +70,7 @@ public interface StreamSourceOperator {
* @param entityPage given entity page
* @return source list response
*/
- PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage);
+ PageInfo<? extends StreamSource> getPageInfo(Page<StreamSourceEntity> entityPage);
/**
* Update the source info.
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java
index b3ea153ee..f95005482 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java
@@ -18,7 +18,6 @@
package org.apache.inlong.manager.service.source;
import com.github.pagehelper.PageInfo;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourcePageRequest;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -68,10 +67,10 @@ public interface StreamSourceService {
/**
* Paging query source information based on conditions.
*
- * @param request Paging request.
- * @return Source info list.
+ * @param request paging request.
+ * @return source list
*/
- PageInfo<? extends SourceListResponse> listByCondition(SourcePageRequest request);
+ PageInfo<? extends StreamSource> listByCondition(SourcePageRequest request);
/**
* Modify data source information
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
index 94220b023..2d7081ce9 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -29,7 +29,6 @@ import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.enums.SourceStatus;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourcePageRequest;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -105,7 +104,7 @@ public class StreamSourceServiceImpl implements StreamSourceService {
throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_NOT_FOUND);
}
StreamSourceOperator operation = operationFactory.getInstance(SourceType.forType(entity.getSourceType()));
- StreamSource streamSource = operation.getByEntity(entity);
+ StreamSource streamSource = operation.getFromEntity(entity);
LOGGER.debug("success to get source by id={}", id);
return streamSource;
}
@@ -132,7 +131,7 @@ public class StreamSourceServiceImpl implements StreamSourceService {
}
@Override
- public PageInfo<? extends SourceListResponse> listByCondition(SourcePageRequest request) {
+ public PageInfo<? extends StreamSource> listByCondition(SourcePageRequest request) {
Preconditions.checkNotNull(request.getInlongGroupId(), ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
PageHelper.startPage(request.getPageNum(), request.getPageSize());
@@ -144,16 +143,16 @@ public class StreamSourceServiceImpl implements StreamSourceService {
SourceType sourceType = SourceType.forType(entity.getSourceType());
sourceMap.computeIfAbsent(sourceType, k -> new Page<>()).add(entity);
}
- List<SourceListResponse> responseList = Lists.newArrayList();
+ List<StreamSource> responseList = Lists.newArrayList();
for (Map.Entry<SourceType, Page<StreamSourceEntity>> entry : sourceMap.entrySet()) {
SourceType sourceType = entry.getKey();
StreamSourceOperator operation = operationFactory.getInstance(sourceType);
- PageInfo<? extends SourceListResponse> pageInfo = operation.getPageInfo(entry.getValue());
+ PageInfo<? extends StreamSource> pageInfo = operation.getPageInfo(entry.getValue());
if (null != pageInfo && CollectionUtils.isNotEmpty(pageInfo.getList())) {
responseList.addAll(pageInfo.getList());
}
}
- PageInfo<? extends SourceListResponse> pageInfo = PageInfo.of(responseList);
+ PageInfo<? extends StreamSource> pageInfo = PageInfo.of(responseList);
LOGGER.debug("success to list source page, result size {}", pageInfo.getSize());
return pageInfo;
@@ -176,7 +175,7 @@ public class StreamSourceServiceImpl implements StreamSourceService {
// Remove id in sourceField when save
List<StreamField> streamFields = request.getFieldList();
if (CollectionUtils.isNotEmpty(streamFields)) {
- streamFields.stream().forEach(streamField -> streamField.setId(null));
+ streamFields.forEach(streamField -> streamField.setId(null));
}
operation.updateOpt(request, groupEntity.getStatus(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/autopush/AutoPushSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/autopush/AutoPushSourceOperator.java
index 5c5d8b588..f6aaa876d 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/autopush/AutoPushSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/autopush/AutoPushSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.autopush;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSource;
import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,7 +34,7 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
* DataProxy SDK source operator
@@ -50,11 +46,13 @@ public class AutoPushSourceOperator extends AbstractSourceOperator {
private ObjectMapper objectMapper;
@Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, AutoPushSourceListResponse::new));
+ public Boolean accept(SourceType sourceType) {
+ return SourceType.AUTO_PUSH == sourceType;
+ }
+
+ @Override
+ protected String getSourceType() {
+ return SourceType.AUTO_PUSH.getType();
}
@Override
@@ -70,32 +68,22 @@ public class AutoPushSourceOperator extends AbstractSourceOperator {
}
@Override
- protected String getSourceType() {
- return SourceType.AUTO_PUSH.getType();
- }
-
- @Override
- protected StreamSource getSource() {
- return new AutoPushSource();
- }
-
- @Override
- public Boolean accept(SourceType sourceType) {
- return SourceType.AUTO_PUSH == sourceType;
- }
-
- @Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ AutoPushSource source = new AutoPushSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
AutoPushSourceDTO dto = AutoPushSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/binlog/BinlogSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/binlog/BinlogSourceOperator.java
index abec6c3bb..ba9e7a824 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/binlog/BinlogSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/binlog/BinlogSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.binlog;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSource;
import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,7 +34,7 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
* Binlog source operator
@@ -59,19 +55,6 @@ public class BinlogSourceOperator extends AbstractSourceOperator {
return SourceType.BINLOG.getType();
}
- @Override
- protected StreamSource getSource() {
- return new MySQLBinlogSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, MySQLBinlogSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
MySQLBinlogSourceRequest sourceRequest = (MySQLBinlogSourceRequest) request;
@@ -85,17 +68,22 @@ public class BinlogSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ MySQLBinlogSource source = new MySQLBinlogSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
MySQLBinlogSourceDTO dto = MySQLBinlogSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java
index fca19a456..14cc47edb 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.file;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
-import org.apache.inlong.manager.common.pojo.source.file.FileSourceDTO;
import org.apache.inlong.manager.common.pojo.source.file.FileSource;
-import org.apache.inlong.manager.common.pojo.source.file.FileSourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.file.FileSourceDTO;
import org.apache.inlong.manager.common.pojo.source.file.FileSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,7 +34,7 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
* File source operator, such as get or set file source info.
@@ -50,17 +46,18 @@ public class FileSourceOperator extends AbstractSourceOperator {
private ObjectMapper objectMapper;
@Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, FileSourceListResponse::new));
+ public Boolean accept(SourceType sourceType) {
+ return SourceType.FILE == sourceType;
+ }
+
+ @Override
+ protected String getSourceType() {
+ return SourceType.SOURCE_FILE;
}
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
FileSourceRequest sourceRequest = (FileSourceRequest) request;
-
try {
CommonBeanUtils.copyProperties(sourceRequest, targetEntity, true);
FileSourceDTO dto = FileSourceDTO.getFromRequest(sourceRequest);
@@ -71,32 +68,22 @@ public class FileSourceOperator extends AbstractSourceOperator {
}
@Override
- protected String getSourceType() {
- return SourceType.SOURCE_FILE;
- }
-
- @Override
- protected StreamSource getSource() {
- return new FileSource();
- }
-
- @Override
- public Boolean accept(SourceType sourceType) {
- return sourceType == SourceType.FILE;
- }
-
- @Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ FileSource source = new FileSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
FileSourceDTO dto = FileSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/kafka/KafkaSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/kafka/KafkaSourceOperator.java
index 5b78edfde..d403d0d89 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/kafka/KafkaSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/kafka/KafkaSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.kafka;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSource;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,10 +34,10 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
- * kafka stream source operator.
+ * kafka stream source operator
*/
@Service
public class KafkaSourceOperator extends AbstractSourceOperator {
@@ -59,19 +55,6 @@ public class KafkaSourceOperator extends AbstractSourceOperator {
return SourceType.KAFKA.getType();
}
- @Override
- protected StreamSource getSource() {
- return new KafkaSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, KafkaSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
KafkaSourceRequest sourceRequest = (KafkaSourceRequest) request;
@@ -85,17 +68,22 @@ public class KafkaSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ KafkaSource source = new KafkaSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
KafkaSourceDTO dto = KafkaSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongo/MongoDBSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongodb/MongoDBSourceOperator.java
similarity index 73%
rename from inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongo/MongoDBSourceOperator.java
rename to inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongodb/MongoDBSourceOperator.java
index 25470a872..d8cfd4883 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongo/MongoDBSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/mongodb/MongoDBSourceOperator.java
@@ -15,22 +15,18 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.source.mongo;
+package org.apache.inlong.manager.service.source.mongodb;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSource;
import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,10 +34,10 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
- * MongoDB source operator.
+ * MongoDB source operator
*/
@Service
public class MongoDBSourceOperator extends AbstractSourceOperator {
@@ -59,19 +55,6 @@ public class MongoDBSourceOperator extends AbstractSourceOperator {
return SourceType.MONGODB.getType();
}
- @Override
- protected StreamSource getSource() {
- return new MongoDBSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, MongoDBSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
MongoDBSourceRequest sourceRequest = (MongoDBSourceRequest) request;
@@ -85,17 +68,22 @@ public class MongoDBSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ MongoDBSource source = new MongoDBSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
MongoDBSourceDTO dto = MongoDBSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/oracle/OracleSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/oracle/OracleSourceOperator.java
index cde6ea93a..f27997578 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/oracle/OracleSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/oracle/OracleSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.oracle;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.oracle.OracleSource;
import org.apache.inlong.manager.common.pojo.source.oracle.OracleSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.oracle.OracleSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.oracle.OracleSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,7 +34,7 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
* Oracle source operator
@@ -59,19 +55,6 @@ public class OracleSourceOperator extends AbstractSourceOperator {
return SourceType.ORACLE.getType();
}
- @Override
- protected StreamSource getSource() {
- return new OracleSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, OracleSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
OracleSourceRequest sourceRequest = (OracleSourceRequest) request;
@@ -85,17 +68,22 @@ public class OracleSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ OracleSource source = new OracleSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
OracleSourceDTO dto = OracleSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgres/PostgresSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgresql/PostgreSQLSourceOperator.java
similarity index 60%
rename from inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgres/PostgresSourceOperator.java
rename to inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgresql/PostgreSQLSourceOperator.java
index dc3d6dc29..40fa73787 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgres/PostgresSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/postgresql/PostgreSQLSourceOperator.java
@@ -15,22 +15,18 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.source.postgres;
+package org.apache.inlong.manager.service.source.postgresql;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
-import org.apache.inlong.manager.common.pojo.source.postgres.PostgresSource;
-import org.apache.inlong.manager.common.pojo.source.postgres.PostgresSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.postgres.PostgresSourceListResponse;
-import org.apache.inlong.manager.common.pojo.source.postgres.PostgresSourceRequest;
+import org.apache.inlong.manager.common.pojo.source.postgresql.PostgreSQLSource;
+import org.apache.inlong.manager.common.pojo.source.postgresql.PostgreSQLSourceDTO;
+import org.apache.inlong.manager.common.pojo.source.postgresql.PostgreSQLSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,13 +34,13 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
- * postgres stream source operator.
+ * PostgreSQL stream source operator
*/
@Service
-public class PostgresSourceOperator extends AbstractSourceOperator {
+public class PostgreSQLSourceOperator extends AbstractSourceOperator {
@Autowired
private ObjectMapper objectMapper;
@@ -59,25 +55,12 @@ public class PostgresSourceOperator extends AbstractSourceOperator {
return SourceType.POSTGRES.getType();
}
- @Override
- protected StreamSource getSource() {
- return new PostgresSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, PostgresSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
- PostgresSourceRequest sourceRequest = (PostgresSourceRequest) request;
+ PostgreSQLSourceRequest sourceRequest = (PostgreSQLSourceRequest) request;
CommonBeanUtils.copyProperties(sourceRequest, targetEntity, true);
try {
- PostgresSourceDTO dto = PostgresSourceDTO.getFromRequest(sourceRequest);
+ PostgreSQLSourceDTO dto = PostgreSQLSourceDTO.getFromRequest(sourceRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
} catch (Exception e) {
throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
@@ -85,17 +68,22 @@ public class PostgresSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ PostgreSQLSource source = new PostgreSQLSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
- PostgresSourceDTO dto = PostgresSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ PostgreSQLSourceDTO dto = PostgreSQLSourceDTO.getFromJson(entity.getExtParams());
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/pulsar/PulsarSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/pulsar/PulsarSourceOperator.java
index baa467682..69b8f9af6 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/pulsar/PulsarSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/pulsar/PulsarSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.pulsar;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
import org.apache.inlong.manager.common.pojo.source.pulsar.PulsarSource;
import org.apache.inlong.manager.common.pojo.source.pulsar.PulsarSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.pulsar.PulsarSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.pulsar.PulsarSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,7 +34,7 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
* Pulsar stream source operator
@@ -49,6 +45,16 @@ public class PulsarSourceOperator extends AbstractSourceOperator {
@Autowired
private ObjectMapper objectMapper;
+ @Override
+ public Boolean accept(SourceType sourceType) {
+ return SourceType.PULSAR == sourceType;
+ }
+
+ @Override
+ protected String getSourceType() {
+ return SourceType.PULSAR.getType();
+ }
+
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
PulsarSourceRequest sourceRequest = (PulsarSourceRequest) request;
@@ -62,40 +68,22 @@ public class PulsarSourceOperator extends AbstractSourceOperator {
}
@Override
- protected String getSourceType() {
- return SourceType.PULSAR.getType();
- }
-
- @Override
- protected StreamSource getSource() {
- return new PulsarSource();
- }
-
- @Override
- public Boolean accept(SourceType sourceType) {
- return SourceType.PULSAR == sourceType;
- }
-
- @Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ PulsarSource source = new PulsarSource();
if (entity == null) {
- return result;
+ return source;
}
+
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
PulsarSourceDTO dto = PulsarSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
- }
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, PulsarSourceListResponse::new));
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SqlServerSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SQLServerSourceOperator.java
similarity index 63%
rename from inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SqlServerSourceOperator.java
rename to inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SQLServerSourceOperator.java
index 6fa9a1c15..7c11c99b4 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SqlServerSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/sqlserver/SQLServerSourceOperator.java
@@ -18,19 +18,15 @@
package org.apache.inlong.manager.service.source.sqlserver;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.pagehelper.Page;
-import com.github.pagehelper.PageInfo;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSource;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSourceDTO;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSourceListResponse;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSourceRequest;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSource;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSourceDTO;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSourceRequest;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
@@ -38,13 +34,13 @@ import org.apache.inlong.manager.service.source.AbstractSourceOperator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.function.Supplier;
+import java.util.List;
/**
- * SqlServer source operator
+ * SQLServer source operator
*/
@Service
-public class SqlServerSourceOperator extends AbstractSourceOperator {
+public class SQLServerSourceOperator extends AbstractSourceOperator {
@Autowired
private ObjectMapper objectMapper;
@@ -59,25 +55,12 @@ public class SqlServerSourceOperator extends AbstractSourceOperator {
return SourceType.SQLSERVER.getType();
}
- @Override
- protected StreamSource getSource() {
- return new SqlServerSource();
- }
-
- @Override
- public PageInfo<? extends SourceListResponse> getPageInfo(Page<StreamSourceEntity> entityPage) {
- if (CollectionUtils.isEmpty(entityPage)) {
- return new PageInfo<>();
- }
- return entityPage.toPageInfo(entity -> this.getFromEntity(entity, SqlServerSourceListResponse::new));
- }
-
@Override
protected void setTargetEntity(SourceRequest request, StreamSourceEntity targetEntity) {
- SqlServerSourceRequest sourceRequest = (SqlServerSourceRequest) request;
+ SQLServerSourceRequest sourceRequest = (SQLServerSourceRequest) request;
CommonBeanUtils.copyProperties(sourceRequest, targetEntity, true);
try {
- SqlServerSourceDTO dto = SqlServerSourceDTO.getFromRequest(sourceRequest);
+ SQLServerSourceDTO dto = SQLServerSourceDTO.getFromRequest(sourceRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
} catch (Exception e) {
throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
@@ -85,17 +68,21 @@ public class SqlServerSourceOperator extends AbstractSourceOperator {
}
@Override
- public <T> T getFromEntity(StreamSourceEntity entity, Supplier<T> target) {
- T result = target.get();
+ public StreamSource getFromEntity(StreamSourceEntity entity) {
+ SQLServerSource source = new SQLServerSource();
if (entity == null) {
- return result;
+ return source;
}
String existType = entity.getSourceType();
Preconditions.checkTrue(getSourceType().equals(existType),
String.format(ErrorCodeEnum.SOURCE_TYPE_NOT_SAME.getMessage(), getSourceType(), existType));
- SqlServerSourceDTO dto = SqlServerSourceDTO.getFromJson(entity.getExtParams());
- CommonBeanUtils.copyProperties(entity, result, true);
- CommonBeanUtils.copyProperties(dto, result, true);
- return result;
+ SQLServerSourceDTO dto = SQLServerSourceDTO.getFromJson(entity.getExtParams());
+ CommonBeanUtils.copyProperties(entity, source, true);
+ CommonBeanUtils.copyProperties(dto, source, true);
+
+ List<StreamField> sourceFields = super.getSourceFields(entity.getId());
+ source.setFieldList(sourceFields);
+ return source;
}
+
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/MongoDBSourceServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/MongoDBSourceServiceTest.java
similarity index 96%
rename from inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/MongoDBSourceServiceTest.java
rename to inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/MongoDBSourceServiceTest.java
index ccf72ea94..719b13379 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/MongoDBSourceServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/MongoDBSourceServiceTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.core.source;
+package org.apache.inlong.manager.service.source;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -24,7 +24,6 @@ import org.apache.inlong.manager.common.pojo.source.mongodb.MongoDBSourceRequest
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.apache.inlong.manager.service.core.impl.InlongStreamServiceTest;
-import org.apache.inlong.manager.service.source.StreamSourceService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/OracleSourceServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/OracleSourceServiceTest.java
similarity index 96%
rename from inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/OracleSourceServiceTest.java
rename to inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/OracleSourceServiceTest.java
index 64281537a..e340ae50b 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/OracleSourceServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/OracleSourceServiceTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.core.source;
+package org.apache.inlong.manager.service.source;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -24,7 +24,6 @@ import org.apache.inlong.manager.common.pojo.source.oracle.OracleSourceRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.apache.inlong.manager.service.core.impl.InlongStreamServiceTest;
-import org.apache.inlong.manager.service.source.StreamSourceService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/SqlServerSourceServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/SQLServerSourceServiceTest.java
similarity index 84%
rename from inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/SqlServerSourceServiceTest.java
rename to inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/SQLServerSourceServiceTest.java
index bd43608df..a1cde4c56 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/SqlServerSourceServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/SQLServerSourceServiceTest.java
@@ -15,16 +15,15 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.core.source;
+package org.apache.inlong.manager.service.source;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSource;
-import org.apache.inlong.manager.common.pojo.source.sqlserver.SqlServerSourceRequest;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSource;
+import org.apache.inlong.manager.common.pojo.source.sqlserver.SQLServerSourceRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.apache.inlong.manager.service.core.impl.InlongStreamServiceTest;
-import org.apache.inlong.manager.service.source.StreamSourceService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -32,7 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired;
/**
* SqlServer source service test.
*/
-public class SqlServerSourceServiceTest extends ServiceBaseTest {
+public class SQLServerSourceServiceTest extends ServiceBaseTest {
private static final String hostname = "127.0.0.1";
private static final Integer port = 1521;
@@ -51,7 +50,7 @@ public class SqlServerSourceServiceTest extends ServiceBaseTest {
public Integer saveSource() {
streamServiceTest.saveInlongStream(GLOBAL_GROUP_ID, GLOBAL_STREAM_ID, GLOBAL_OPERATOR);
- SqlServerSourceRequest sourceInfo = new SqlServerSourceRequest();
+ SQLServerSourceRequest sourceInfo = new SQLServerSourceRequest();
sourceInfo.setInlongGroupId(GLOBAL_GROUP_ID);
sourceInfo.setInlongStreamId(GLOBAL_STREAM_ID);
sourceInfo.setSourceName(sourceName);
@@ -88,8 +87,8 @@ public class SqlServerSourceServiceTest extends ServiceBaseTest {
StreamSource response = sourceService.get(id);
Assertions.assertEquals(GLOBAL_GROUP_ID, response.getInlongGroupId());
- SqlServerSource sqlserverSource = (SqlServerSource) response;
- SqlServerSourceRequest request = CommonBeanUtils.copyProperties(sqlserverSource, SqlServerSourceRequest::new);
+ SQLServerSource sqlserverSource = (SQLServerSource) response;
+ SQLServerSourceRequest request = CommonBeanUtils.copyProperties(sqlserverSource, SQLServerSourceRequest::new);
boolean result = sourceService.update(request, GLOBAL_OPERATOR);
Assertions.assertTrue(result);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/StreamSourceServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/StreamSourceServiceTest.java
similarity index 96%
rename from inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/StreamSourceServiceTest.java
rename to inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/StreamSourceServiceTest.java
index 1aa8aca14..056c55043 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/source/StreamSourceServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/StreamSourceServiceTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.service.core.source;
+package org.apache.inlong.manager.service.source;
import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -24,7 +24,6 @@ import org.apache.inlong.manager.common.pojo.source.mysql.MySQLBinlogSourceReque
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.apache.inlong.manager.service.core.impl.InlongStreamServiceTest;
-import org.apache.inlong.manager.service.source.StreamSourceService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
index 1fad7d979..76cfe838f 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
@@ -24,7 +24,6 @@ import io.swagger.annotations.ApiOperation;
import org.apache.inlong.manager.common.beans.Response;
import org.apache.inlong.manager.common.enums.OperationType;
import org.apache.inlong.manager.common.pojo.common.UpdateValidation;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourcePageRequest;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.StreamSource;
@@ -65,7 +64,7 @@ public class StreamSourceController {
@RequestMapping(value = "/source/list", method = RequestMethod.GET)
@ApiOperation(value = "Get stream source list by paginating")
- public Response<PageInfo<? extends SourceListResponse>> listByCondition(SourcePageRequest request) {
+ public Response<PageInfo<? extends StreamSource>> listByCondition(SourcePageRequest request) {
return Response.success(sourceService.listByCondition(request));
}
diff --git a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/UtilsTest.java b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/UtilsTest.java
index b54cd927a..54da0bbb4 100644
--- a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/UtilsTest.java
+++ b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/UtilsTest.java
@@ -18,6 +18,7 @@
package org.apache.inlong.manager.web;
import org.apache.inlong.manager.common.pojo.group.InlongGroupRequest;
+import org.apache.inlong.manager.common.pojo.group.pulsar.InlongPulsarRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
import org.junit.jupiter.api.Assertions;
@@ -45,7 +46,7 @@ public class UtilsTest {
entity.setCreateTime(new Date());
entity.setModifyTime(null);
- InlongGroupRequest request = new InlongGroupRequest();
+ InlongGroupRequest request = new InlongPulsarRequest();
request.setInlongGroupId("info");
BeanUtils.copyProperties(request, entity);
@@ -64,7 +65,7 @@ public class UtilsTest {
entity.setCreateTime(new Date());
entity.setModifyTime(null);
- InlongGroupRequest request = new InlongGroupRequest();
+ InlongGroupRequest request = new InlongPulsarRequest();
request.setInlongGroupId("info");
CommonBeanUtils.copyProperties(request, entity, true);