You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by zi...@apache.org on 2022/03/25 02:51:37 UTC
[incubator-inlong] branch master updated: [INLONG-3355][Manager] Check the source states before return group state in manager client
This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new c0078df [INLONG-3355][Manager] Check the source states before return group state in manager client
c0078df is described below
commit c0078dfc1db6bdd71791ba769ea405311c1c303f
Author: kipshi <48...@users.noreply.github.com>
AuthorDate: Fri Mar 25 10:51:30 2022 +0800
[INLONG-3355][Manager] Check the source states before return group state in manager client
---
.../manager/client/api/InlongGroupContext.java | 35 ++++++++++++++++++++--
.../api/impl/DefaultInlongStreamBuilder.java | 10 +++----
.../manager/client/api/impl/InlongStreamImpl.java | 4 +--
.../client/api/inner/InnerInlongManagerClient.java | 12 ++++----
.../client/api/inner/InnerStreamContext.java | 6 ++--
.../manager/client/api/util/InlongParser.java | 6 ++--
.../client/api/util/InlongStreamSinkTransfer.java | 10 +++----
.../api/util/InlongStreamSourceTransfer.java | 8 ++---
.../client/api/util/InlongStreamTransfer.java | 8 ++---
.../common/pojo/stream/FullStreamResponse.java | 2 +-
...ngStreamResponse.java => InlongStreamInfo.java} | 7 ++++-
.../common/pojo/stream/InlongStreamResponse.java | 17 ++++++-----
.../workflow/form/GroupResourceProcessForm.java | 4 +--
.../pojo/workflow/form/UpdateGroupProcessForm.java | 4 +--
.../mappers/InlongStreamFieldEntityMapper.xml | 4 ++-
.../manager/service/CommonOperateService.java | 4 +--
.../manager/service/core/InlongStreamService.java | 4 +--
.../core/impl/InlongGroupProcessOperation.java | 6 ++--
.../service/core/impl/InlongStreamServiceImpl.java | 10 +++----
.../source/AbstractStreamSourceOperation.java | 6 ----
.../listener/AbstractSourceOperateListener.java | 2 +-
.../thirdparty/sort/PushSortConfigListener.java | 4 +--
.../thirdparty/sort/util/SerializationUtils.java | 8 ++---
.../thirdparty/sort/util/SourceInfoUtils.java | 6 ++--
.../listener/StartCreateGroupProcessListener.java | 8 ++---
.../service/core/impl/InlongStreamServiceTest.java | 4 +--
.../source/listener/DataSourceListenerTest.java | 4 +--
.../thirdparty/sort/DisableZkForSortTest.java | 10 +++----
.../service/workflow/WorkflowServiceImplTest.java | 4 +--
.../web/controller/InlongStreamController.java | 2 +-
30 files changed, 126 insertions(+), 93 deletions(-)
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
index 92aaf76..70ee2eb 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
@@ -20,6 +20,7 @@ package org.apache.inlong.manager.client.api;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.inlong.manager.client.api.StreamSource.State;
@@ -35,6 +36,7 @@ import java.util.List;
import java.util.Map;
@Data
+@Slf4j
public class InlongGroupContext implements Serializable {
private String groupId;
@@ -91,24 +93,53 @@ public class InlongGroupContext implements Serializable {
if (MapUtils.isEmpty(this.inlongStreamMap)) {
return;
}
+ List<StreamSource> sourcesInGroup = Lists.newArrayList();
List<StreamSource> failedSources = Lists.newArrayList();
this.inlongStreamMap.values().stream().forEach(inlongStream -> {
Map<String, StreamSource> sources = inlongStream.getSources();
if (MapUtils.isNotEmpty(sources)) {
for (Map.Entry<String, StreamSource> entry : sources.entrySet()) {
StreamSource source = entry.getValue();
- if (source.getState() == State.FAILED) {
- failedSources.add(source);
+ if (source != null) {
+ sourcesInGroup.add(source);
+ if (source.getState() == State.FAILED) {
+ failedSources.add(source);
+ }
}
}
}
});
+ // check if any stream source is failed
if (CollectionUtils.isNotEmpty(failedSources)) {
this.state = InlongGroupState.FAILED;
for (StreamSource failedSource : failedSources) {
this.groupErrLogs.computeIfAbsent("failedSources", Lists::newArrayList)
.add(GsonUtil.toJson(failedSource));
}
+ return;
+ }
+ // check if any stream source is in indirect state
+ switch (this.state) {
+ case STARTED:
+ for (StreamSource source : sourcesInGroup) {
+ if (source.getState() != State.NORMAL) {
+ log.warn("StreamSource:{} is not started", source);
+ this.state = InlongGroupState.INITIALIZING;
+ break;
+ }
+ }
+ return;
+ case STOPPED:
+ for (StreamSource source : sourcesInGroup) {
+ if (source.getState() != State.FROZEN) {
+ log.warn("StreamSource:{} is not stopped", source);
+ this.state = InlongGroupState.OPERATING;
+ break;
+ }
+ }
+ return;
+ default:
+ return;
}
}
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 a391f5b..cd248ff 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
@@ -42,7 +42,7 @@ 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.stream.InlongStreamFieldInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.List;
@@ -68,7 +68,7 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
if (MapUtils.isEmpty(groupContext.getStreamContextMap())) {
groupContext.setStreamContextMap(Maps.newHashMap());
}
- InlongStreamResponse stream = InlongStreamTransfer.createStreamInfo(streamConf, groupContext.getGroupInfo());
+ InlongStreamInfo stream = InlongStreamTransfer.createStreamInfo(streamConf, groupContext.getGroupInfo());
InnerStreamContext streamContext = new InnerStreamContext(stream);
groupContext.setStreamContext(streamContext);
this.streamContext = streamContext;
@@ -107,7 +107,7 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
@Override
public InlongStream init() {
- InlongStreamResponse streamInfo = streamContext.getStreamInfo();
+ InlongStreamInfo streamInfo = streamContext.getStreamInfo();
String streamIndex = managerClient.createStreamInfo(streamInfo);
streamInfo.setId(Double.valueOf(streamIndex).intValue());
//Create source and update index
@@ -127,8 +127,8 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
@Override
public InlongStream initOrUpdate() {
- InlongStreamResponse dataStreamInfo = streamContext.getStreamInfo();
- Pair<Boolean, InlongStreamResponse> existMsg = managerClient.isStreamExists(dataStreamInfo);
+ InlongStreamInfo dataStreamInfo = streamContext.getStreamInfo();
+ Pair<Boolean, InlongStreamInfo> existMsg = managerClient.isStreamExists(dataStreamInfo);
if (existMsg.getKey()) {
Pair<Boolean, String> updateMsg = managerClient.updateStreamInfo(dataStreamInfo);
if (updateMsg.getKey()) {
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 d6a2ad7..9ed62f3 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
@@ -35,7 +35,7 @@ import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
import org.apache.inlong.manager.common.pojo.stream.FullStreamResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.List;
import java.util.Map;
@@ -55,7 +55,7 @@ public class InlongStreamImpl extends InlongStream {
private List<StreamField> streamFields = Lists.newArrayList();
public InlongStreamImpl(FullStreamResponse fullStreamResponse) {
- InlongStreamResponse streamInfo = fullStreamResponse.getStreamInfo();
+ InlongStreamInfo streamInfo = fullStreamResponse.getStreamInfo();
this.name = streamInfo.getName();
List<InlongStreamFieldInfo> streamFieldInfos = streamInfo.getFieldList();
if (CollectionUtils.isNotEmpty(streamFieldInfos)) {
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 3d298a7..5495385 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
@@ -51,7 +51,7 @@ import org.apache.inlong.manager.common.pojo.stream.FullStreamResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamApproveRequest;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamConfigLogListResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.EventLogView;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
import org.apache.inlong.manager.common.util.JsonUtils;
@@ -282,7 +282,7 @@ public class InnerInlongManagerClient {
}
}
- public String createStreamInfo(InlongStreamResponse streamInfo) {
+ public String createStreamInfo(InlongStreamInfo streamInfo) {
String path = HTTP_PATH + "/stream/save";
final String stream = GsonUtil.toJson(streamInfo);
final RequestBody streamBody = RequestBody.create(MediaType.parse("application/json"), stream);
@@ -307,8 +307,8 @@ public class InnerInlongManagerClient {
}
}
- public Pair<Boolean, InlongStreamResponse> isStreamExists(InlongStreamResponse streamInfo) {
- InlongStreamResponse currentStreamInfo = getStreamInfo(streamInfo);
+ public Pair<Boolean, InlongStreamInfo> isStreamExists(InlongStreamInfo streamInfo) {
+ InlongStreamInfo currentStreamInfo = getStreamInfo(streamInfo);
if (currentStreamInfo != null) {
return Pair.of(true, currentStreamInfo);
} else {
@@ -316,7 +316,7 @@ public class InnerInlongManagerClient {
}
}
- public Pair<Boolean, String> updateStreamInfo(InlongStreamResponse streamInfo) {
+ public Pair<Boolean, String> updateStreamInfo(InlongStreamInfo streamInfo) {
streamInfo.setCreateTime(null);
streamInfo.setModifyTime(null);
final String path = HTTP_PATH + "/stream/update";
@@ -344,7 +344,7 @@ public class InnerInlongManagerClient {
}
}
- public InlongStreamResponse getStreamInfo(InlongStreamResponse streamInfo) {
+ public InlongStreamInfo getStreamInfo(InlongStreamInfo streamInfo) {
String path = HTTP_PATH + "/stream/get";
String url = formatUrl(path);
url += String.format("&groupId=%s&streamId=%s", streamInfo.getInlongGroupId(), streamInfo.getInlongStreamId());
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerStreamContext.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerStreamContext.java
index 1dc3d32..5b85758 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerStreamContext.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerStreamContext.java
@@ -23,7 +23,7 @@ import lombok.NoArgsConstructor;
import org.apache.inlong.manager.common.pojo.sink.SinkRequest;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.List;
import java.util.Map;
@@ -32,13 +32,13 @@ import java.util.Map;
@NoArgsConstructor
public class InnerStreamContext {
- private InlongStreamResponse streamInfo;
+ private InlongStreamInfo streamInfo;
private Map<String, SourceRequest> sourceRequests = Maps.newHashMap();
private Map<String, SinkRequest> sinkRequests = Maps.newHashMap();
- public InnerStreamContext(InlongStreamResponse streamInfo) {
+ public InnerStreamContext(InlongStreamInfo streamInfo) {
this.streamInfo = streamInfo;
}
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
index c608960..4a61ca7 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
@@ -46,7 +46,7 @@ import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceResponse;
import org.apache.inlong.manager.common.pojo.stream.FullStreamResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamApproveRequest;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamConfigLogListResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.EventLogView;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
@@ -103,9 +103,9 @@ public class InlongParser {
}.getType());
}
- public static InlongStreamResponse parseStreamInfo(Response response) {
+ public static InlongStreamInfo parseStreamInfo(Response response) {
Object data = response.getData();
- return GsonUtil.fromJson(GsonUtil.toJson(data), InlongStreamResponse.class);
+ return GsonUtil.fromJson(GsonUtil.toJson(data), InlongStreamInfo.class);
}
public static List<FullStreamResponse> parseStreamList(Response response) {
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
index 38bf6e7..5920b17 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
@@ -40,7 +40,7 @@ import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkRequest;
import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkResponse;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkRequest;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import java.nio.charset.Charset;
@@ -49,7 +49,7 @@ import java.util.stream.Collectors;
public class InlongStreamSinkTransfer {
- public static SinkRequest createSinkRequest(StreamSink streamSink, InlongStreamResponse streamInfo) {
+ public static SinkRequest createSinkRequest(StreamSink streamSink, InlongStreamInfo streamInfo) {
SinkType sinkType = streamSink.getSinkType();
SinkRequest sinkRequest;
if (sinkType == SinkType.HIVE) {
@@ -84,7 +84,7 @@ public class InlongStreamSinkTransfer {
return streamSinkResult;
}
- private static SinkRequest createClickHouseRequest(StreamSink streamSink, InlongStreamResponse streamInfo) {
+ private static SinkRequest createClickHouseRequest(StreamSink streamSink, InlongStreamInfo streamInfo) {
ClickHouseSinkRequest clickHouseSinkRequest = new ClickHouseSinkRequest();
ClickHouseSink clickHouseSink = (ClickHouseSink) streamSink;
clickHouseSinkRequest.setSinkName(clickHouseSink.getSinkName());
@@ -157,7 +157,7 @@ public class InlongStreamSinkTransfer {
}
- private static SinkRequest createKafkaRequest(StreamSink streamSink, InlongStreamResponse streamInfo) {
+ private static SinkRequest createKafkaRequest(StreamSink streamSink, InlongStreamInfo streamInfo) {
KafkaSinkRequest kafkaSinkRequest = new KafkaSinkRequest();
KafkaSink kafkaSink = (KafkaSink) streamSink;
kafkaSinkRequest.setSinkName(streamSink.getSinkName());
@@ -200,7 +200,7 @@ public class InlongStreamSinkTransfer {
return kafkaSink;
}
- private static HiveSinkRequest createHiveRequest(StreamSink streamSink, InlongStreamResponse streamInfo) {
+ private static HiveSinkRequest createHiveRequest(StreamSink streamSink, InlongStreamInfo streamInfo) {
HiveSinkRequest hiveSinkRequest = new HiveSinkRequest();
HiveSink hiveSink = (HiveSink) streamSink;
hiveSinkRequest.setSinkName(streamSink.getSinkName());
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
index f90d995..115eb37 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
@@ -38,7 +38,7 @@ import org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceResponse;
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.source.kafka.KafkaSourceResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.Arrays;
@@ -47,7 +47,7 @@ import java.util.Arrays;
*/
public class InlongStreamSourceTransfer {
- public static SourceRequest createSourceRequest(StreamSource streamSource, InlongStreamResponse streamInfo) {
+ public static SourceRequest createSourceRequest(StreamSource streamSource, InlongStreamInfo streamInfo) {
SourceType sourceType = streamSource.getSourceType();
switch (sourceType) {
case KAFKA:
@@ -176,7 +176,7 @@ public class InlongStreamSourceTransfer {
return binlogSource;
}
- private static KafkaSourceRequest createKafkaSourceRequest(KafkaSource kafkaSource, InlongStreamResponse stream) {
+ private static KafkaSourceRequest createKafkaSourceRequest(KafkaSource kafkaSource, InlongStreamInfo stream) {
KafkaSourceRequest sourceRequest = new KafkaSourceRequest();
sourceRequest.setSourceName(kafkaSource.getSourceName());
sourceRequest.setInlongGroupId(stream.getInlongGroupId());
@@ -199,7 +199,7 @@ public class InlongStreamSourceTransfer {
}
private static BinlogSourceRequest createBinlogSourceRequest(MySQLBinlogSource binlogSource,
- InlongStreamResponse streamInfo) {
+ InlongStreamInfo streamInfo) {
BinlogSourceRequest sourceRequest = new BinlogSourceRequest();
sourceRequest.setSourceName(binlogSource.getSourceName());
sourceRequest.setInlongGroupId(streamInfo.getInlongGroupId());
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
index bf68845..28a6301 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
@@ -23,15 +23,15 @@ import org.apache.inlong.manager.client.api.InlongStreamConf;
import org.apache.inlong.manager.client.api.StreamField;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.List;
import java.util.stream.Collectors;
public class InlongStreamTransfer {
- public static InlongStreamResponse createStreamInfo(InlongStreamConf streamConf, InlongGroupInfo groupInfo) {
- InlongStreamResponse dataStreamInfo = new InlongStreamResponse();
+ public static InlongStreamInfo createStreamInfo(InlongStreamConf streamConf, InlongGroupInfo groupInfo) {
+ InlongStreamInfo dataStreamInfo = new InlongStreamInfo();
dataStreamInfo.setInlongGroupId(groupInfo.getInlongGroupId());
final String streamId = "b_" + streamConf.getName();
dataStreamInfo.setInlongStreamId(streamId);
@@ -57,7 +57,7 @@ public class InlongStreamTransfer {
}
public static List<InlongStreamFieldInfo> createStreamFields(
- List<StreamField> fieldList, InlongStreamResponse streamInfo) {
+ List<StreamField> fieldList, InlongStreamInfo streamInfo) {
return fieldList.stream().map(field -> {
InlongStreamFieldInfo fieldInfo = new InlongStreamFieldInfo();
fieldInfo.setInlongStreamId(streamInfo.getInlongStreamId());
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/FullStreamResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/FullStreamResponse.java
index 8fca538..1248130 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/FullStreamResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/FullStreamResponse.java
@@ -33,7 +33,7 @@ import java.util.List;
public class FullStreamResponse {
@ApiModelProperty("Inlong stream information")
- private InlongStreamResponse streamInfo;
+ private InlongStreamInfo streamInfo;
@ApiModelProperty("Stream source information")
private List<SourceResponse> sourceInfo;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
similarity index 93%
copy from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
copy to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
index 8eb80db..37be4a4 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
@@ -22,6 +22,7 @@ import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
import java.util.Date;
import java.util.List;
@@ -32,7 +33,7 @@ import java.util.List;
@Data
@EqualsAndHashCode(callSuper = true)
@ApiModel("Inlong stream info")
-public class InlongStreamResponse extends InlongStreamBaseInfo {
+public class InlongStreamInfo extends InlongStreamBaseInfo {
@ApiModelProperty(value = "Primary key")
private Integer id;
@@ -110,4 +111,8 @@ public class InlongStreamResponse extends InlongStreamBaseInfo {
@ApiModelProperty(value = "Field list")
private List<InlongStreamFieldInfo> fieldList;
+ public InlongStreamResponse genResponse() {
+ return CommonBeanUtils.copyProperties(this, InlongStreamResponse::new);
+ }
+
}
\ No newline at end of file
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
index 8eb80db..0c8b875 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
@@ -20,19 +20,20 @@ package org.apache.inlong.manager.common.pojo.stream;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
-import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
import java.util.Date;
import java.util.List;
-/**
- * Inlong stream info
- */
@Data
-@EqualsAndHashCode(callSuper = true)
-@ApiModel("Inlong stream info")
-public class InlongStreamResponse extends InlongStreamBaseInfo {
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel("Inlong stream response")
+public class InlongStreamResponse {
@ApiModelProperty(value = "Primary key")
private Integer id;
@@ -110,4 +111,4 @@ public class InlongStreamResponse extends InlongStreamBaseInfo {
@ApiModelProperty(value = "Field list")
private List<InlongStreamFieldInfo> fieldList;
-}
\ No newline at end of file
+}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/GroupResourceProcessForm.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/GroupResourceProcessForm.java
index b2c8f99..ef7dd15 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/GroupResourceProcessForm.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/GroupResourceProcessForm.java
@@ -22,7 +22,7 @@ import lombok.Data;
import lombok.EqualsAndHashCode;
import org.apache.inlong.manager.common.exceptions.FormValidateException;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import java.util.List;
import java.util.Map;
@@ -40,7 +40,7 @@ public class GroupResourceProcessForm extends BaseProcessForm {
private String streamId;
- private List<InlongStreamResponse> streamList;
+ private List<InlongStreamInfo> inlongStreamInfoList;
public InlongGroupInfo getGroupInfo() {
return groupInfo;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/UpdateGroupProcessForm.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/UpdateGroupProcessForm.java
index a367a7f..a25ccfd 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/UpdateGroupProcessForm.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/UpdateGroupProcessForm.java
@@ -25,7 +25,7 @@ import lombok.Getter;
import lombok.Setter;
import org.apache.inlong.manager.common.exceptions.FormValidateException;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.util.Preconditions;
import java.util.List;
@@ -44,7 +44,7 @@ public class UpdateGroupProcessForm extends BaseProcessForm {
@ApiModelProperty(value = "OperateType to define the update operation", required = true)
private OperateType operateType;
- private List<InlongStreamResponse> streamList;
+ private List<InlongStreamInfo> streamList;
@Override
public void validate() throws FormValidateException {
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamFieldEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamFieldEntityMapper.xml
index 08f1afb..93ff35c 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamFieldEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamFieldEntityMapper.xml
@@ -61,7 +61,7 @@
(
#{item.id}, #{item.inlongGroupId}, #{item.inlongStreamId}, #{item.isPredefinedField},
#{item.fieldName}, #{item.fieldValue}, #{item.preExpression}, #{item.fieldType},
- #{item.fieldComment}, #{isMetaField}, #{item.rankNum}, #{item.isDeleted}
+ #{item.fieldComment}, #{item.isMetaField}, #{item.rankNum}, #{item.isDeleted}
)
</foreach>
ON DUPLICATE KEY UPDATE
@@ -73,6 +73,7 @@
field_value = values(field_value),
pre_expression = values(pre_expression),
field_type = values(field_type),
+ is_meta_field = values(is_meta_field),
field_comment = values(field_comment),
rank_num = values(rank_num),
is_deleted = values(is_deleted)
@@ -114,6 +115,7 @@
field_value = #{fieldValue,jdbcType=VARCHAR},
pre_expression = #{preExpression,jdbcType=VARCHAR},
field_type = #{fieldType,jdbcType=VARCHAR},
+ is_meta_field = #(isMetaField,jdbcType=SMALLINT),
field_comment = #{fieldComment,jdbcType=VARCHAR},
rank_num = #{rankNum,jdbcType=SMALLINT},
is_deleted = #{isDeleted,jdbcType=INTEGER}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/CommonOperateService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/CommonOperateService.java
index 362aaff..eaa314f 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/CommonOperateService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/CommonOperateService.java
@@ -33,7 +33,7 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkResponse;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.util.JsonUtils;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
@@ -224,7 +224,7 @@ public class CommonOperateService {
boolean isAllMigration = SourceInfoUtils.isBinlogAllMigration(sourceResponse);
List<FieldMappingUnit> mappingUnitList;
- InlongStreamResponse streamInfo = streamService.get(groupId, streamId);
+ InlongStreamInfo streamInfo = streamService.get(groupId, streamId);
if (isAllMigration) {
mappingUnitList = FieldInfoUtils.setAllMigrationFieldMapping(sourceFields, sinkFields);
} else {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
index 5388a7d..0487dc1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/InlongStreamService.java
@@ -21,7 +21,7 @@ import com.github.pagehelper.PageInfo;
import org.apache.inlong.manager.common.pojo.stream.FullStreamRequest;
import org.apache.inlong.manager.common.pojo.stream.FullStreamResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamApproveRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamListResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamRequest;
@@ -54,7 +54,7 @@ public interface InlongStreamService {
* @param streamId Inlong stream id
* @return inlong stream details
*/
- InlongStreamResponse get(String groupId, String streamId);
+ InlongStreamInfo get(String groupId, String streamId);
/**
* Query whether the inlong stream ID exists
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperation.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperation.java
index 70dab08..b9aa88c 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperation.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperation.java
@@ -20,7 +20,7 @@ package org.apache.inlong.manager.service.core.impl;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.inlong.manager.common.enums.GroupState;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.StreamBriefResponse;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
import org.apache.inlong.manager.common.pojo.workflow.form.NewGroupProcessForm;
@@ -193,8 +193,8 @@ public class InlongGroupProcessOperation {
if (OperateType.RESTART == operateType) {
List<InlongStreamEntity> inlongStreamEntityList =
streamMapper.selectByGroupId(groupInfo.getInlongGroupId());
- List<InlongStreamResponse> streamList = CommonBeanUtils.copyListProperties(inlongStreamEntityList,
- InlongStreamResponse::new);
+ List<InlongStreamInfo> streamList = CommonBeanUtils.copyListProperties(inlongStreamEntityList,
+ InlongStreamInfo::new);
form.setStreamList(streamList);
}
form.setGroupInfo(groupInfo);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
index 268dc78..263b1e8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
@@ -41,7 +41,7 @@ import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamListResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamTopicResponse;
import org.apache.inlong.manager.common.pojo.stream.StreamBriefResponse;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
@@ -133,7 +133,7 @@ public class InlongStreamServiceImpl implements InlongStreamService {
}
@Override
- public InlongStreamResponse get(String groupId, String streamId) {
+ public InlongStreamInfo get(String groupId, String streamId) {
LOGGER.debug("begin to get inlong stream by groupId={}, streamId={}", groupId, streamId);
Preconditions.checkNotNull(groupId, Constant.GROUP_ID_IS_EMPTY);
Preconditions.checkNotNull(streamId, Constant.STREAM_ID_IS_EMPTY);
@@ -144,7 +144,7 @@ public class InlongStreamServiceImpl implements InlongStreamService {
throw new BusinessException(ErrorCodeEnum.STREAM_NOT_FOUND);
}
- InlongStreamResponse response = CommonBeanUtils.copyProperties(streamEntity, InlongStreamResponse::new);
+ InlongStreamInfo response = CommonBeanUtils.copyProperties(streamEntity, InlongStreamInfo::new);
List<InlongStreamFieldInfo> streamFields = this.getStreamFields(groupId, streamId);
response.setFieldList(streamFields);
@@ -443,11 +443,11 @@ public class InlongStreamServiceImpl implements InlongStreamService {
PageHelper.startPage(request.getPageNum(), request.getPageSize());
Page<InlongStreamEntity> page = (Page<InlongStreamEntity>) streamMapper.selectByCondition(request);
- List<InlongStreamResponse> streamInfoList = CommonBeanUtils.copyListProperties(page, InlongStreamResponse::new);
+ List<InlongStreamInfo> streamInfoList = CommonBeanUtils.copyListProperties(page, InlongStreamInfo::new);
// Convert and encapsulate the paged results
List<FullStreamResponse> responseList = new ArrayList<>(streamInfoList.size());
- for (InlongStreamResponse streamInfo : streamInfoList) {
+ for (InlongStreamInfo streamInfo : streamInfoList) {
// 2Set the field information of the inlong stream
String streamId = streamInfo.getInlongStreamId();
List<InlongStreamFieldInfo> streamFields = getStreamFields(groupId, streamId);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractStreamSourceOperation.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractStreamSourceOperation.java
index c3eafd9..be68d6b 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractStreamSourceOperation.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractStreamSourceOperation.java
@@ -122,15 +122,9 @@ public abstract class AbstractStreamSourceOperation implements StreamSourceOpera
throw new RuntimeException(String.format("Source=%s is not allowed to update, "
+ "please wait until its changed to final status or stop / frozen / delete it firstly", entity));
}
-
// Setting updated parameters of stream source entity.
setTargetEntity(request, entity);
entity.setVersion(entity.getVersion() + 1);
- if (GroupState.forCode(groupStatus).equals(GroupState.CONFIG_SUCCESSFUL)) {
- entity.setStatus(SourceState.TO_BE_ISSUED_ADD.getCode());
- } else {
- entity.setStatus(SourceState.SOURCE_NEW.getCode());
- }
entity.setModifier(operator);
entity.setModifyTime(new Date());
sourceMapper.updateByPrimaryKeySelective(entity);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
index ac02145..f8232e9 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
@@ -66,7 +66,7 @@ public abstract class AbstractSourceOperateListener implements DataSourceOperate
@Override
public ListenerResult listen(WorkflowContext context) throws Exception {
- log.info("Delete data source for context={}", context);
+ log.info("Operate data source for context={}", context);
InlongGroupInfo groupInfo = getGroupInfo(context.getProcessForm());
final String groupId = groupInfo.getInlongGroupId();
List<StreamBriefResponse> streamBriefResponses = streamService.getBriefList(groupId);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/PushSortConfigListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/PushSortConfigListener.java
index 0d5f45b..16ccb9d 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/PushSortConfigListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/PushSortConfigListener.java
@@ -23,7 +23,7 @@ import org.apache.inlong.manager.common.enums.Constant;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
import org.apache.inlong.manager.common.util.JsonUtils;
import org.apache.inlong.manager.dao.entity.StreamSinkFieldEntity;
@@ -125,7 +125,7 @@ public class PushSortConfigListener implements SortOperateListener {
List<StreamSinkFieldEntity> fieldList) {
DeserializationInfo deserializationInfo = null;
String groupId = groupInfo.getInlongGroupId();
- InlongStreamResponse streamInfo = streamService.get(groupId, streamId);
+ InlongStreamInfo streamInfo = streamService.get(groupId, streamId);
boolean isDbType = Constant.DATA_SOURCE_DB.equals(streamInfo.getDataType());
if (!isDbType) {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SerializationUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SerializationUtils.java
index a2211b0..b4d4a52 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SerializationUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SerializationUtils.java
@@ -25,7 +25,7 @@ import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkResponse;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
import org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceResponse;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.sort.protocol.deserialization.AvroDeserializationInfo;
import org.apache.inlong.sort.protocol.deserialization.CanalDeserializationInfo;
import org.apache.inlong.sort.protocol.deserialization.CsvDeserializationInfo;
@@ -48,7 +48,7 @@ public class SerializationUtils {
* Create deserialization info
*/
public static DeserializationInfo createDeserialInfo(SourceResponse sourceResponse,
- InlongStreamResponse streamInfo) {
+ InlongStreamInfo streamInfo) {
SourceType sourceType = SourceType.forType(sourceResponse.getSourceType());
switch (sourceType) {
case BINLOG:
@@ -87,7 +87,7 @@ public class SerializationUtils {
/**
* Get deserialization info for Kafka
*/
- private static DeserializationInfo deserializeForKafka(KafkaSourceResponse source, InlongStreamResponse stream) {
+ private static DeserializationInfo deserializeForKafka(KafkaSourceResponse source, InlongStreamInfo stream) {
String serializationType = source.getSerializationType();
DataTypeEnum dataType = DataTypeEnum.forName(serializationType);
switch (dataType) {
@@ -136,7 +136,7 @@ public class SerializationUtils {
* Get deserialization info for File
*/
private static DeserializationInfo deserializeForFile(SourceResponse sourceResponse,
- InlongStreamResponse streamInfo) {
+ InlongStreamInfo streamInfo) {
String serializationType = sourceResponse.getSerializationType();
DataTypeEnum dataType = DataTypeEnum.forName(serializationType);
switch (dataType) {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SourceInfoUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SourceInfoUtils.java
index 66838ff..13a74c8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SourceInfoUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdparty/sort/util/SourceInfoUtils.java
@@ -27,7 +27,7 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPulsarInfo;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
import org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceResponse;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.deserialization.DeserializationInfo;
@@ -61,7 +61,7 @@ public class SourceInfoUtils {
* Create source info for DataFlowInfo.
*/
public static SourceInfo createSourceInfo(PulsarClusterInfo pulsarCluster, String masterAddress,
- ClusterBean clusterBean, InlongGroupInfo groupInfo, InlongStreamResponse streamInfo,
+ ClusterBean clusterBean, InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
SourceResponse sourceResponse, List<FieldInfo> sourceFields) {
String mqType = groupInfo.getMiddlewareType();
@@ -84,7 +84,7 @@ public class SourceInfoUtils {
* Create source info for Pulsar
*/
private static SourceInfo createPulsarSourceInfo(PulsarClusterInfo pulsarCluster, ClusterBean clusterBean,
- InlongGroupInfo groupInfo, InlongStreamResponse streamInfo,
+ InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
DeserializationInfo deserializationInfo, List<FieldInfo> fieldInfos) {
String topicName = streamInfo.getMqResourceObj();
InlongGroupPulsarInfo pulsarInfo = (InlongGroupPulsarInfo) groupInfo.getMqExtInfo();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/listener/StartCreateGroupProcessListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/listener/StartCreateGroupProcessListener.java
index 37b5e30..4bf4903 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/listener/StartCreateGroupProcessListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/listener/StartCreateGroupProcessListener.java
@@ -19,7 +19,7 @@ package org.apache.inlong.manager.service.workflow.group.listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
import org.apache.inlong.manager.common.pojo.workflow.form.NewGroupProcessForm;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
@@ -69,9 +69,9 @@ public class StartCreateGroupProcessListener implements ProcessEventListener {
processForm.setGroupInfo(groupService.get(groupId));
String username = context.getApplicant();
List<InlongStreamEntity> inlongStreamEntityList = streamMapper.selectByGroupId(groupId);
- List<InlongStreamResponse> streamList = CommonBeanUtils.copyListProperties(inlongStreamEntityList,
- InlongStreamResponse::new);
- processForm.setStreamList(streamList);
+ List<InlongStreamInfo> streamList = CommonBeanUtils.copyListProperties(inlongStreamEntityList,
+ InlongStreamInfo::new);
+ processForm.setInlongStreamInfoList(streamList);
workflowService.start(ProcessName.CREATE_GROUP_RESOURCE, username, processForm);
return ListenerResult.success();
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceTest.java
index 88b98a8..7a6884e 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceTest.java
@@ -18,7 +18,7 @@
package org.apache.inlong.manager.service.core.impl;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.service.core.InlongStreamService;
import org.junit.Assert;
import org.junit.Test;
@@ -47,7 +47,7 @@ public class InlongStreamServiceTest {
public Integer saveInlongStream(String groupId, String streamId, String operator) {
;
try {
- InlongStreamResponse response = streamService.get(groupId, streamId);
+ InlongStreamInfo response = streamService.get(groupId, streamId);
if (response != null) {
return response.getId();
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/DataSourceListenerTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/DataSourceListenerTest.java
index 367baed..4791146 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/DataSourceListenerTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/DataSourceListenerTest.java
@@ -24,7 +24,7 @@ import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
import org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.ProcessResponse;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
@@ -51,7 +51,7 @@ public class DataSourceListenerTest extends WorkflowServiceImplTest {
private StreamSourceService streamSourceService;
public Integer createBinlogSource(InlongGroupInfo groupInfo) {
- final InlongStreamResponse stream = createStreamInfo(groupInfo);
+ final InlongStreamInfo stream = createStreamInfo(groupInfo);
BinlogSourceRequest sourceRequest = new BinlogSourceRequest();
sourceRequest.setInlongGroupId(stream.getInlongGroupId());
sourceRequest.setInlongStreamId(stream.getInlongStreamId());
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/thirdparty/sort/DisableZkForSortTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/thirdparty/sort/DisableZkForSortTest.java
index 653a56b..cdf2ffd 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/thirdparty/sort/DisableZkForSortTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/thirdparty/sort/DisableZkForSortTest.java
@@ -24,7 +24,7 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.sink.SinkFieldRequest;
import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkRequest;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.ProcessResponse;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
@@ -63,7 +63,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
@Autowired
protected StreamSourceService streamSourceService;
- public HiveSinkRequest createHiveSink(InlongStreamResponse streamInfo) {
+ public HiveSinkRequest createHiveSink(InlongStreamInfo streamInfo) {
HiveSinkRequest hiveSinkRequest = new HiveSinkRequest();
hiveSinkRequest.setInlongGroupId(streamInfo.getInlongGroupId());
hiveSinkRequest.setSinkType("HIVE");
@@ -95,7 +95,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
return hiveSinkRequest;
}
- public KafkaSourceRequest createKafkaSource(InlongStreamResponse streamInfo) {
+ public KafkaSourceRequest createKafkaSource(InlongStreamInfo streamInfo) {
KafkaSourceRequest kafkaSourceRequest = new KafkaSourceRequest();
kafkaSourceRequest.setInlongGroupId(streamInfo.getInlongGroupId());
kafkaSourceRequest.setInlongStreamId(streamInfo.getInlongStreamId());
@@ -112,7 +112,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
groupInfo.setStatus(GroupState.CONFIG_SUCCESSFUL.getCode());
groupInfo.setZookeeperEnabled(0);
groupService.update(groupInfo.genRequest(), OPERATOR);
- InlongStreamResponse streamInfo = createStreamInfo(groupInfo);
+ InlongStreamInfo streamInfo = createStreamInfo(groupInfo);
createHiveSink(streamInfo);
createKafkaSource(streamInfo);
mockTaskListenerFactory();
@@ -139,7 +139,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
groupInfo.setZookeeperEnabled(0);
groupInfo.setStatus(GroupState.CONFIG_SUCCESSFUL.getCode());
groupService.update(groupInfo.genRequest(), OPERATOR);
- InlongStreamResponse streamInfo = createStreamInfo(groupInfo);
+ InlongStreamInfo streamInfo = createStreamInfo(groupInfo);
createHiveSink(streamInfo);
createKafkaSource(streamInfo);
UpdateGroupProcessForm form = new UpdateGroupProcessForm();
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
index 9cbb07a..ebec742 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
@@ -26,7 +26,7 @@ import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPulsarInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamRequest;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.common.pojo.workflow.ProcessResponse;
import org.apache.inlong.manager.common.pojo.workflow.TaskExecuteLogQuery;
import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
@@ -149,7 +149,7 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
/**
* Create inlong stream
*/
- public InlongStreamResponse createStreamInfo(InlongGroupInfo groupInfo) {
+ public InlongStreamInfo createStreamInfo(InlongGroupInfo groupInfo) {
// delete first
try {
streamService.delete(GROUP_ID, OPERATOR, OPERATOR);
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
index 1454268..430dd2a 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongStreamController.java
@@ -85,7 +85,7 @@ public class InlongStreamController {
@ApiImplicitParam(name = "streamId", dataTypeClass = String.class, required = true)
})
public Response<InlongStreamResponse> get(@RequestParam String groupId, @RequestParam String streamId) {
- return Response.success(streamService.get(groupId, streamId));
+ return Response.success(streamService.get(groupId, streamId).genResponse());
}
@RequestMapping(value = "/list", method = RequestMethod.POST)