You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by qh...@apache.org on 2015/06/11 03:44:15 UTC
incubator-kylin git commit: add log to track Exception when failed to
fetch response from Kafka
Repository: incubator-kylin
Updated Branches:
refs/heads/0.8.0 3a31d916a -> 006e68e37
add log to track Exception when failed to fetch response from Kafka
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/006e68e3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/006e68e3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/006e68e3
Branch: refs/heads/0.8.0
Commit: 006e68e37aa78708b99bf852272fda80b4335cf1
Parents: 3a31d91
Author: qianhao.zhou <qi...@ebay.com>
Authored: Thu Jun 11 09:44:01 2015 +0800
Committer: qianhao.zhou <qi...@ebay.com>
Committed: Thu Jun 11 09:44:01 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/job/streaming/StreamingBootstrap.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/006e68e3/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 5bdc670..11b9127 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
@@ -237,18 +237,20 @@ public class StreamingBootstrap {
int clusterId = 0;
final ExecutorService executorService = Executors.newFixedThreadPool(10);
+ final long targetTimestamp = startTimestamp - margin;
for (final KafkaClusterConfig kafkaClusterConfig : streamingConfig.getKafkaClusterConfigs()) {
final ConcurrentMap<Integer, Long> partitionIdOffsetMap = Maps.newConcurrentMap();
final int partitionCount = KafkaRequester.getKafkaTopicMeta(kafkaClusterConfig).getPartitionIds().size();
final CountDownLatch countDownLatch = new CountDownLatch(partitionCount);
for (int i = 0; i < partitionCount; ++i) {
final int idx = i;
- final long start = startTimestamp - margin;
- executorService.submit(new Runnable() {
+ executorService.execute(new Runnable() {
@Override
public void run() {
try {
- partitionIdOffsetMap.put(idx, StreamingUtil.findClosestOffsetWithDataTimestamp(kafkaClusterConfig, idx, start, streamParser));
+ partitionIdOffsetMap.put(idx, StreamingUtil.findClosestOffsetWithDataTimestamp(kafkaClusterConfig, idx, targetTimestamp, streamParser));
+ } catch (Exception e) {
+ logger.error(String.format("fail to get start offset partitionId: %d, target timestamp: %d", idx, targetTimestamp));
} finally {
countDownLatch.countDown();
}