[inlong] branch master updated: [INLONG-5536][Manger] Fix version missing when updating stream info (#5537)

[dockerzhang]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang 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 92c34e9b9 [INLONG-5536][Manger] Fix version missing when updating 
stream info (#5537)
92c34e9b9 is described below

commit 92c34e9b90bffff1ef532d1515a3289d44a6019d
Author: woofyzhao <490467...@qq.com>
AuthorDate: Sun Aug 14 17:38:10 2022 +0800

    [INLONG-5536][Manger] Fix version missing when updating stream info (#5537)
---
 .../manager/client/api/impl/DefaultInlongStreamBuilder.java |  5 +++--
 .../manager/client/api/inner/client/InlongStreamClient.java | 13 +++++++++++++
 2 files changed, 16 insertions(+), 2 deletions(-)

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 b3d16e4bb..6077795ba 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
@@ -156,8 +156,9 @@ public class DefaultInlongStreamBuilder extends 
InlongStreamBuilder {
         InlongStreamInfo dataStreamInfo = streamContext.getStreamInfo();
         StreamPipeline streamPipeline = inlongStream.createPipeline();
         dataStreamInfo.setExtParams(JsonUtils.toJsonString(streamPipeline));
-        Boolean isExist = streamClient.isStreamExists(dataStreamInfo);
-        if (isExist) {
+        InlongStreamInfo existStreamInfo = 
streamClient.getStreamIfExists(dataStreamInfo);
+        if (existStreamInfo != null) {
+            dataStreamInfo.setVersion(existStreamInfo.getVersion());
             Pair<Boolean, String> updateMsg = 
streamClient.updateStreamInfo(dataStreamInfo);
             if (!updateMsg.getKey()) {
                 throw new RuntimeException(String.format("Update data stream 
failed:%s", updateMsg.getValue()));
diff --git 
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InlongStreamClient.java
 
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InlongStreamClient.java
index ad24f03bc..924296d1d 100644
--- 
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InlongStreamClient.java
+++ 
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InlongStreamClient.java
@@ -99,6 +99,19 @@ public class InlongStreamClient {
         }
     }
 
+    /**
+     * Get inlong stream by the given inlong group id and stream id.
+     *
+     * @param streamInfo the given inlong stream info
+     * @return inlong stream info if exists, null will be returned if not exits
+     */
+    public InlongStreamInfo getStreamIfExists(InlongStreamInfo streamInfo) {
+        if (this.isStreamExists(streamInfo)) {
+            return getStreamInfo(streamInfo.getInlongGroupId(), 
streamInfo.getInlongStreamId());
+        }
+        return null;
+    }
+
     /**
      * Paging query inlong stream brief info list
      *