[inlong] branch master updated: [INLONG-6371][Manager] Fix the lack of topic and cache zone properties in getSortSource (#6372)

[healchow]

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 f5d597e6f [INLONG-6371][Manager] Fix the lack of topic and cache zone 
properties in getSortSource (#6372)
f5d597e6f is described below

commit f5d597e6ffe6fb8da4b5398ddc4d3e2d7a8d97de
Author: vernedeng <deng...@pku.edu.cn>
AuthorDate: Wed Nov 2 18:33:48 2022 +0800

    [INLONG-6371][Manager] Fix the lack of topic and cache zone properties in 
getSortSource (#6372)
---
 .../inlong/manager/service/core/impl/SortSourceServiceImpl.java     | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortSourceServiceImpl.java
 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortSourceServiceImpl.java
index 8be2d31df..4ec62c36d 100644
--- 
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortSourceServiceImpl.java
+++ 
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SortSourceServiceImpl.java
@@ -364,7 +364,10 @@ public class SortSourceServiceImpl implements 
SortSourceService {
                         topic = 
Optional.ofNullable(backupStreamMqResource.get(info.getGroupId())).orElse(topic);
                     }
                     String fullTopic = 
tenant.concat("/").concat(namespace).concat("/").concat(topic);
-                    return Topic.builder().topic(fullTopic).build();
+                    return Topic.builder()
+                            .topic(fullTopic)
+                            .topicProperties(info.getExtParamsMap())
+                            .build();
                 })
                 .collect(Collectors.toList());
         return CacheZone.builder()
@@ -372,6 +375,7 @@ public class SortSourceServiceImpl implements 
SortSourceService {
                 .serviceUrl(cluster.getUrl())
                 .topics(sdkTopics)
                 .authentication(auth)
+                .cacheZoneProperties(cluster.getExtParamsMap())
                 .zoneType(ClusterType.PULSAR)
                 .build();
     }