[kylin] branch master updated: KYLIN-3752 Increase broadcaster's concurrency to avoid exceptions

[shaofengshi]

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

shaofengshi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kylin.git


The following commit(s) were added to refs/heads/master by this push:
     new 250f1a1  KYLIN-3752 Increase broadcaster's concurrency to avoid 
exceptions
250f1a1 is described below

commit 250f1a1941432059e86207fbf483f183d5b3ed9a
Author: xbirbird <31679225+xbirb...@users.noreply.github.com>
AuthorDate: Wed Jan 2 10:43:19 2019 +0800

    KYLIN-3752 Increase broadcaster's concurrency to avoid exceptions
---
 .../java/org/apache/kylin/metadata/cachesync/Broadcaster.java    | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git 
a/core-metadata/src/main/java/org/apache/kylin/metadata/cachesync/Broadcaster.java
 
b/core-metadata/src/main/java/org/apache/kylin/metadata/cachesync/Broadcaster.java
index b9c03b6..834401c 100644
--- 
a/core-metadata/src/main/java/org/apache/kylin/metadata/cachesync/Broadcaster.java
+++ 
b/core-metadata/src/main/java/org/apache/kylin/metadata/cachesync/Broadcaster.java
@@ -91,14 +91,17 @@ public class Broadcaster {
         this.config = config;
         this.syncErrorHandler = getSyncErrorHandler(config);
         this.announceMainLoop = Executors.newSingleThreadExecutor(new 
DaemonThreadFactory());
-        this.announceThreadPool = new ThreadPoolExecutor(1, 10, 60L, 
TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>(), new 
DaemonThreadFactory());
-
+        
         final String[] nodes = config.getRestServers();
         if (nodes == null || nodes.length < 1) {
             logger.warn("There is no available rest server; check the 
'kylin.server.cluster-servers' config");
         }
         logger.debug("{} nodes in the cluster: {}", (nodes == null ? 0 : 
nodes.length), Arrays.toString(nodes));
+        
+        int corePoolSize = (nodes == null || nodes.length < 1)? 1 : 
nodes.length;
+        int maximumPoolSize = (nodes == null || nodes.length < 1)? 10 : 
nodes.length * 2;
+        this.announceThreadPool = new ThreadPoolExecutor(corePoolSize, 
maximumPoolSize, 60L, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<Runnable>(), new DaemonThreadFactory());
 
         announceMainLoop.execute(new Runnable() {
             @Override