[INLONG-10597][Sort] Provide default pulsar producer configuration

vernedeng · Jul 10, 2024 · b290ec9 · b290ec9
1 parent 841bc02
commit b290ec9
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 9 deletions.
diff --git a/.../main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java b/.../main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java
@@ -76,11 +76,6 @@ public void reload() {
                 this.pulsarNodeConfig = requestNodeConfig;
             }
 
-            CacheClusterConfig clusterConfig = new CacheClusterConfig();
-            clusterConfig.setClusterName(this.taskName);
-            clusterConfig.setParams(this.sortTaskConfig.getSinkParams());
-            this.cacheClusterConfig = clusterConfig;
-
             this.taskConfig = newTaskConfig;
             this.sortTaskConfig = newSortTaskConfig;
 

diff --git a/...ce/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java b/...ce/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java
@@ -24,7 +24,6 @@
 import org.apache.inlong.sort.standalone.utils.Constants;
 import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
 
-import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flume.Context;
 import org.apache.flume.Transaction;
@@ -165,8 +164,7 @@ private void initBuilderByCacheCluster(CacheClusterConfig cacheClusterConfig) {
 
     private void initBuilderByNodeConfig(PulsarNodeConfig nodeConfig) {
         this.cacheClusterName = nodeConfig.getNodeName();
-        this.context = new Context(nodeConfig.getProperties() == null ?
-                new HashMap<>() : nodeConfig.getProperties());
+        this.context = new Context(nodeConfig.getProperties() == null ? new HashMap<>() : nodeConfig.getProperties());
 
         clientBuilder = PulsarClient.builder();
         String serviceUrl = nodeConfig.getServiceUrl();
@@ -310,5 +308,4 @@ public boolean send(ProfileEvent profileEvent, Transaction tx) throws IOExceptio
         return true;
     }
 
-
 }