kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From qhz...@apache.org
Subject incubator-kylin git commit: optimize OneOffStreamBuilder 1. replace Executors.newFixedThreadPool with newCachedThreadPool 2. retry when fail to get message from Kafka Producer, not just throw Exception
Date Fri, 03 Jul 2015 03:00:57 GMT
Repository: incubator-kylin
Updated Branches:
  refs/heads/0.8 6d920a9ff -> a88e10bf2


optimize OneOffStreamBuilder
1. replace Executors.newFixedThreadPool with newCachedThreadPool
2. retry when fail to get message from Kafka Producer, not just throw
Exception


Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/a88e10bf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/a88e10bf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/a88e10bf

Branch: refs/heads/0.8
Commit: a88e10bf2030de73abf78e098d8a3b4b8f5369c4
Parents: 6d920a9
Author: qianhao.zhou <qianzhou@ebay.com>
Authored: Fri Jul 3 10:57:27 2015 +0800
Committer: qianhao.zhou <qianzhou@ebay.com>
Committed: Fri Jul 3 10:57:27 2015 +0800

----------------------------------------------------------------------
 .../kylin/job/streaming/StreamingBootstrap.java       |  5 ++++-
 .../org/apache/kylin/streaming/StreamingUtil.java     | 14 ++++++++++----
 2 files changed, 14 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/a88e10bf/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java b/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
index d4c5139..aaa4aa4 100644
--- a/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
+++ b/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
@@ -236,7 +236,10 @@ public class StreamingBootstrap {
         final List<BlockingQueue<StreamMessage>> queues = Lists.newLinkedList();
 
         int clusterId = 0;
-        final ExecutorService executorService = Executors.newFixedThreadPool(10, new DaemonThreadFactory());
+        final ExecutorService executorService = new ThreadPoolExecutor(10, 10,
+                15L, TimeUnit.SECONDS,
+                new SynchronousQueue<Runnable>(),
+                new DaemonThreadFactory());
         final long targetTimestamp = startTimestamp - margin;
         for (final KafkaClusterConfig kafkaClusterConfig : streamingConfig.getKafkaClusterConfigs())
{
             final ConcurrentMap<Integer, Long> partitionIdOffsetMap = Maps.newConcurrentMap();

http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/a88e10bf/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
index 264c384..961c725 100644
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
+++ b/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
@@ -36,15 +36,21 @@ public final class StreamingUtil {
     private static MessageAndOffset getKafkaMessage(KafkaClusterConfig kafkaClusterConfig,
int partitionId, long offset) {
         final String topic = kafkaClusterConfig.getTopic();
         int retry = 3;
-        while (retry > 0) {
-            final Broker leadBroker = Preconditions.checkNotNull(getLeadBroker(kafkaClusterConfig,
partitionId), "unable to find leadBroker with config:" + kafkaClusterConfig + " partitionId:"
+ partitionId);
+        while (retry-- > 0) {
+            final Broker leadBroker = getLeadBroker(kafkaClusterConfig, partitionId);
+            if (leadBroker == null) {
+                logger.warn("unable to find leadBroker with config:" + kafkaClusterConfig
+ " partitionId:" + partitionId);
+                continue;
+            }
             final FetchResponse response = KafkaRequester.fetchResponse(topic, partitionId,
offset, leadBroker, kafkaClusterConfig);
-            Preconditions.checkArgument(response.errorCode(topic, partitionId) == 0, "errorCode
of FetchResponse is:" + response.errorCode(topic, partitionId));
+            if (response.errorCode(topic, partitionId) != 0) {
+                logger.warn("errorCode of FetchResponse is:" + response.errorCode(topic,
partitionId));
+                continue;
+            }
             final Iterator<MessageAndOffset> iterator = response.messageSet(topic,
partitionId).iterator();
             if (iterator.hasNext()) {
                 return iterator.next();
             } else {
-                retry--;
                 try {
                     Thread.sleep(1000);
                 } catch (InterruptedException e) {


Mime
View raw message