(inlong) branch master updated: [INLONG-9373][Manager] Fix the issue of failed creation of pulsar namespace (#9374)

[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 94ce7a5350 [INLONG-9373][Manager] Fix the issue of failed creation of 
pulsar namespace (#9374)
94ce7a5350 is described below

commit 94ce7a5350d8f13c6dcea26ee3bea8686f50b202
Author: fuweng11 <76141879+fuwen...@users.noreply.github.com>
AuthorDate: Fri Dec 1 12:35:34 2023 +0800

    [INLONG-9373][Manager] Fix the issue of failed creation of pulsar namespace 
(#9374)
---
 .../org/apache/inlong/manager/common/consts/SinkType.java    |  2 +-
 .../manager/service/operationlog/OperationLogRecorder.java   |  5 ++++-
 .../service/resource/queue/pulsar/PulsarOperator.java        | 12 ++++++------
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git 
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/SinkType.java
 
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/SinkType.java
index 1d53c71fc2..6169426932 100644
--- 
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/SinkType.java
+++ 
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/SinkType.java
@@ -38,7 +38,7 @@ public class SinkType extends StreamType {
     public static final String HBASE = "HBASE";
 
     @SupportSortType(sortType = SortType.SORT_STANDALONE)
-    public static final String ELASTICSEARCH = "ELASTICSEARCH";
+    public static final String ELASTICSEARCH = "ES";
 
     @SupportSortType(sortType = SortType.SORT_FLINK)
     public static final String HDFS = "HDFS";
diff --git 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/operationlog/OperationLogRecorder.java
 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/operationlog/OperationLogRecorder.java
index b7dc2a0443..99a1e9472a 100644
--- 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/operationlog/OperationLogRecorder.java
+++ 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/operationlog/OperationLogRecorder.java
@@ -77,9 +77,12 @@ public class OperationLogRecorder {
             try {
                 JSONObject obj = (JSONObject) JSON.toJSON(arg);
                 for (String key : obj.keySet()) {
-                    if (Objects.equals(key, INLONG_GROUP_ID) || 
Objects.equals(key, INLONG_STREAM_ID)) {
+                    if (Objects.equals(key, INLONG_GROUP_ID)) {
                         groupId = obj.getString(key);
                     }
+                    if (Objects.equals(key, INLONG_STREAM_ID)) {
+                        streamId = obj.getString(key);
+                    }
                 }
             } catch (Exception ignored) {
                 log.debug("do nothing when exception");
diff --git 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
index 7c55f65e43..5ff2c9cecd 100644
--- 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
+++ 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
@@ -108,13 +108,13 @@ public class PulsarOperator {
         Preconditions.expectNotBlank(namespace, 
ErrorCodeEnum.INVALID_PARAMETER,
                 "pulsar namespace cannot be empty during create namespace");
 
-        String namespaceName = tenant + "/" + namespace;
-        LOGGER.info("begin to create namespace={}", namespaceName);
+        String tenantNamespaceName = tenant + "/" + namespace;
+        LOGGER.info("begin to create namespace={}", tenantNamespaceName);
         try {
             // Check whether the namespace exists, and create it if it does 
not exist
             boolean isExists = this.namespaceExists(pulsarClusterInfo, tenant, 
namespace);
             if (isExists) {
-                LOGGER.warn("namespace={} already exists, skip to create", 
namespaceName);
+                LOGGER.warn("namespace={} already exists, skip to create", 
tenantNamespaceName);
                 return;
             }
 
@@ -147,10 +147,10 @@ public class PulsarOperator {
                     pulsarInfo.getWriteQuorum(), pulsarInfo.getAckQuorum(), 
pulsarInfo.getMaxMarkDeleteRate());
             policies.setPersistence(persistencePolicies);
 
-            PulsarUtils.createNamespace(restTemplate, pulsarClusterInfo, 
tenant, namespaceName, policies);
-            LOGGER.info("success to create namespace={}", namespaceName);
+            PulsarUtils.createNamespace(restTemplate, pulsarClusterInfo, 
tenant, namespace, policies);
+            LOGGER.info("success to create namespace={}", tenantNamespaceName);
         } catch (Exception e) {
-            LOGGER.error("failed to create namespace=" + namespaceName, e);
+            LOGGER.error("failed to create namespace=" + tenantNamespaceName, 
e);
             throw e;
         }
     }