You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2015/06/08 05:25:16 UTC
incubator-kylin git commit: KYLIN-808 bug fix in MicroBatch union
Repository: incubator-kylin
Updated Branches:
refs/heads/0.8.0 c04847264 -> d4e5120f3
KYLIN-808 bug fix in MicroBatch union
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/d4e5120f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/d4e5120f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/d4e5120f
Branch: refs/heads/0.8.0
Commit: d4e5120f3b721d66052acf7b196958efbccad00f
Parents: c048472
Author: shaofengshi <sh...@apache.org>
Authored: Mon Jun 8 11:24:23 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Mon Jun 8 11:25:07 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/job/streaming/StreamingBootstrap.java | 7 ++++++-
.../java/org/apache/kylin/streaming/MicroStreamBatch.java | 4 ++--
.../main/java/org/apache/kylin/streaming/StreamBuilder.java | 3 ++-
3 files changed, 10 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d4e5120f/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 d22e7eb..bcc1b97 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
@@ -146,7 +146,12 @@ public class StreamingBootstrap {
if (partitionIdOffsetMap.containsKey(transferredPartitionId)) {
final long earliestOffset = KafkaRequester.getLastOffset(kafkaClusterConfig.getTopic(), partitionId, OffsetRequest.EarliestTime(), leadBroker, kafkaClusterConfig);
long committedOffset = partitionIdOffsetMap.get(transferredPartitionId);
- Preconditions.checkArgument(committedOffset >= earliestOffset && committedOffset <= latestOffset, String.format("invalid offset:%d, earliestOffset:%d, latestOffset:%d", committedOffset, earliestOffset, latestOffset));
+ Preconditions.checkArgument(committedOffset <= latestOffset, String.format("invalid offset:%d, earliestOffset:%d, latestOffset:%d", committedOffset, earliestOffset, latestOffset));
+
+ if (committedOffset < earliestOffset) {
+ logger.warn(String.format("invalid offset:%d, earliestOffset:%d; Will use earliestOffset as committedOffset.", committedOffset, earliestOffset));
+ committedOffset = earliestOffset;
+ }
streamingOffset = committedOffset;
}
logger.info("starting offset:" + streamingOffset + " cluster id:" + clusterID + " partitionId:" + partitionId + " transferredPartitionId:" + transferredPartitionId);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d4e5120f/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java b/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
index 6ecd592..c5e563c 100644
--- a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
+++ b/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
@@ -82,9 +82,9 @@ public final class MicroStreamBatch {
MicroStreamBatch result = new MicroStreamBatch(one);
result.streams.addAll(another.streams);
result.offset.setFirst(Math.min(result.offset.getFirst(), another.offset.getFirst()));
- result.offset.setSecond(Math.min(result.offset.getSecond(), another.offset.getSecond()));
+ result.offset.setSecond(Math.max(result.offset.getSecond(), another.offset.getSecond()));
result.timestamp.setFirst(Math.min(result.timestamp.getFirst(), another.timestamp.getFirst()));
- result.timestamp.setSecond(Math.min(result.timestamp.getSecond(), another.timestamp.getSecond()));
+ result.timestamp.setSecond(Math.max(result.timestamp.getSecond(), another.timestamp.getSecond()));
result.rawMessageCount = one.rawMessageCount + another.rawMessageCount;
return result;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/d4e5120f/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
index 3da4d79..76e58cf 100644
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
+++ b/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
@@ -152,7 +152,8 @@ public class StreamBuilder implements Runnable {
MicroStreamBatch batch = batches.get(0);
if (batches.size() > 1) {
for (int i = 1; i < inputCount; i++) {
- batch = MicroStreamBatch.union(batch, batches.get(i));
+ if (batches.get(i).size() > 0)
+ batch = MicroStreamBatch.union(batch, batches.get(i));
}
}
if (batchInterval > 0) {