You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2016/02/05 06:48:29 UTC
[1/3] storm git commit: STORM-1455: Do not reset the emittedOffset
for offsetOutOfRangeExceptions
Repository: storm
Updated Branches:
refs/heads/1.x-branch 65a025305 -> 7090be454
STORM-1455: Do not reset the emittedOffset for offsetOutOfRangeExceptions
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0f5abf43
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0f5abf43
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0f5abf43
Branch: refs/heads/1.x-branch
Commit: 0f5abf43dfd02eb878b5ec09b098c224ba8c5bc0
Parents: 65a0253
Author: Abhishek Agarwal <ab...@inmobi.com>
Authored: Tue Jan 19 21:43:22 2016 +0530
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Feb 5 14:39:32 2016 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/kafka/KafkaUtils.java | 44 +++++++++++---------
.../apache/storm/kafka/PartitionManager.java | 42 +++++++++++++------
2 files changed, 54 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0f5abf43/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
index 8cd0fd0..a2be825 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
@@ -17,24 +17,15 @@
*/
package org.apache.storm.kafka;
-import org.apache.storm.metric.api.IMetric;
-import org.apache.storm.utils.Utils;
import com.google.common.base.Preconditions;
-import kafka.api.FetchRequest;
-import kafka.api.FetchRequestBuilder;
-import kafka.api.PartitionOffsetRequestInfo;
-import kafka.common.TopicAndPartition;
-import kafka.javaapi.FetchResponse;
-import kafka.javaapi.OffsetRequest;
-import kafka.javaapi.consumer.SimpleConsumer;
-import kafka.javaapi.message.ByteBufferMessageSet;
-import kafka.message.Message;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
import org.apache.storm.kafka.trident.GlobalPartitionInformation;
import org.apache.storm.kafka.trident.IBrokerReader;
import org.apache.storm.kafka.trident.StaticBrokerReader;
import org.apache.storm.kafka.trident.ZkBrokerReader;
+import org.apache.storm.metric.api.IMetric;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.ConnectException;
@@ -43,6 +34,16 @@ import java.nio.ByteBuffer;
import java.nio.channels.UnresolvedAddressException;
import java.util.*;
+import kafka.api.FetchRequest;
+import kafka.api.FetchRequestBuilder;
+import kafka.api.PartitionOffsetRequestInfo;
+import kafka.common.TopicAndPartition;
+import kafka.javaapi.FetchResponse;
+import kafka.javaapi.OffsetRequest;
+import kafka.javaapi.consumer.SimpleConsumer;
+import kafka.javaapi.message.ByteBufferMessageSet;
+import kafka.message.Message;
+
public class KafkaUtils {
@@ -80,7 +81,7 @@ public class KafkaUtils {
}
public static class KafkaOffsetMetric implements IMetric {
- Map<Partition, Long> _partitionToOffset = new HashMap<Partition, Long>();
+ Map<Partition, PartitionManager.OffsetData> _partitionToOffset = new HashMap<Partition, PartitionManager.OffsetData>();
Set<Partition> _partitions;
DynamicPartitionConnections _connections;
@@ -88,8 +89,8 @@ public class KafkaUtils {
_connections = connections;
}
- public void setLatestEmittedOffset(Partition partition, long offset) {
- _partitionToOffset.put(partition, offset);
+ public void setOffsetData(Partition partition, PartitionManager.OffsetData offsetData) {
+ _partitionToOffset.put(partition, offsetData);
}
private class TopicMetrics {
@@ -97,6 +98,7 @@ public class KafkaUtils {
long totalEarliestTimeOffset = 0;
long totalLatestTimeOffset = 0;
long totalLatestEmittedOffset = 0;
+ long totalLatestCompletedOffset = 0;
}
@Override
@@ -105,7 +107,7 @@ public class KafkaUtils {
HashMap ret = new HashMap();
if (_partitions != null && _partitions.size() == _partitionToOffset.size()) {
Map<String,TopicMetrics> topicMetricsMap = new TreeMap<String, TopicMetrics>();
- for (Map.Entry<Partition, Long> e : _partitionToOffset.entrySet()) {
+ for (Map.Entry<Partition, PartitionManager.OffsetData> e : _partitionToOffset.entrySet()) {
Partition partition = e.getKey();
SimpleConsumer consumer = _connections.getConnection(partition);
if (consumer == null) {
@@ -118,8 +120,9 @@ public class KafkaUtils {
LOG.warn("No data found in Kafka Partition " + partition.getId());
return null;
}
- long latestEmittedOffset = e.getValue();
- long spoutLag = latestTimeOffset - latestEmittedOffset;
+ long latestEmittedOffset = e.getValue().latestEmittedOffset;
+ long latestCompletedOffset = e.getValue().latestCompletedOffset;
+ long spoutLag = latestTimeOffset - latestCompletedOffset;
String topic = partition.topic;
String metricPath = partition.getId();
//Handle the case where Partition Path Id does not contain topic name Partition.getId() == "partition_" + partition
@@ -130,6 +133,7 @@ public class KafkaUtils {
ret.put(metricPath + "/" + "earliestTimeOffset", earliestTimeOffset);
ret.put(metricPath + "/" + "latestTimeOffset", latestTimeOffset);
ret.put(metricPath + "/" + "latestEmittedOffset", latestEmittedOffset);
+ ret.put(metricPath + "/" + "latestCompletedOffset", latestCompletedOffset);
if (!topicMetricsMap.containsKey(partition.topic)) {
topicMetricsMap.put(partition.topic,new TopicMetrics());
@@ -140,6 +144,7 @@ public class KafkaUtils {
topicMetrics.totalEarliestTimeOffset += earliestTimeOffset;
topicMetrics.totalLatestTimeOffset += latestTimeOffset;
topicMetrics.totalLatestEmittedOffset += latestEmittedOffset;
+ topicMetrics.totalLatestCompletedOffset += latestCompletedOffset;
}
for(Map.Entry<String, TopicMetrics> e : topicMetricsMap.entrySet()) {
@@ -149,6 +154,7 @@ public class KafkaUtils {
ret.put(topic + "/" + "totalEarliestTimeOffset", topicMetrics.totalEarliestTimeOffset);
ret.put(topic + "/" + "totalLatestTimeOffset", topicMetrics.totalLatestTimeOffset);
ret.put(topic + "/" + "totalLatestEmittedOffset", topicMetrics.totalLatestEmittedOffset);
+ ret.put(topic + "/" + "totalLatestCompletedOffset", topicMetrics.totalLatestCompletedOffset);
}
return ret;
http://git-wip-us.apache.org/repos/asf/storm/blob/0f5abf43/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
index 0f5aaa0..bcc4001 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
@@ -17,28 +17,26 @@
*/
package org.apache.storm.kafka;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+
import org.apache.storm.Config;
+import org.apache.storm.kafka.KafkaSpout.EmitState;
+import org.apache.storm.kafka.trident.MaxMetric;
import org.apache.storm.metric.api.CombinedMetric;
import org.apache.storm.metric.api.CountMetric;
import org.apache.storm.metric.api.MeanReducer;
import org.apache.storm.metric.api.ReducedMetric;
import org.apache.storm.spout.SpoutOutputCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableMap;
+import java.util.*;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.storm.kafka.KafkaSpout.EmitState;
-import org.apache.storm.kafka.trident.MaxMetric;
-
-import java.util.*;
-
public class PartitionManager {
public static final Logger LOG = LoggerFactory.getLogger(PartitionManager.class);
@@ -186,8 +184,7 @@ public class PartitionManager {
try {
msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset);
} catch (TopicOffsetOutOfRangeException e) {
- _emittedToOffset = KafkaUtils.getOffset(_consumer, _partition.topic, _partition.partition, kafka.api.OffsetRequest.EarliestTime());
- LOG.warn("{} Using new offset: {}", _partition.partition, _emittedToOffset);
+ offset = KafkaUtils.getOffset(_consumer, _partition.topic, _partition.partition, kafka.api.OffsetRequest.EarliestTime());
// fetch failed, so don't update the metrics
//fix bug [STORM-643] : remove outdated failed offsets
@@ -196,10 +193,15 @@ public class PartitionManager {
// all the failed offsets, that are earlier than actual EarliestTime
// offset, since they are anyway not there.
// These calls to broker API will be then saved.
- Set<Long> omitted = this._failedMsgRetryManager.clearInvalidMessages(_emittedToOffset);
+ Set<Long> omitted = this._failedMsgRetryManager.clearInvalidMessages(offset);
LOG.warn("Removing the failed offsets that are out of range: {}", omitted);
}
+
+ if (offset > _emittedToOffset) {
+ _emittedToOffset = offset;
+ LOG.warn("{} Using new offset: {}", _partition.partition, _emittedToOffset);
+ }
return;
}
@@ -294,6 +296,10 @@ public class PartitionManager {
}
}
+ public OffsetData getOffsetData() {
+ return new OffsetData(_emittedToOffset, lastCompletedOffset());
+ }
+
public Partition getPartition() {
return _partition;
}
@@ -313,4 +319,14 @@ public class PartitionManager {
this.offset = offset;
}
}
+
+ static class OffsetData {
+ public long latestEmittedOffset;
+ public long latestCompletedOffset;
+
+ public OffsetData(long latestEmittedOffset, long latestCompletedOffset) {
+ this.latestEmittedOffset = latestEmittedOffset;
+ this.latestCompletedOffset = latestCompletedOffset;
+ }
+ }
}
[2/3] storm git commit: STORM-1455: Fix build failure
Posted by ka...@apache.org.
STORM-1455: Fix build failure
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5eac6855
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5eac6855
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5eac6855
Branch: refs/heads/1.x-branch
Commit: 5eac6855642df52823af9ba13ad12ab5505f9753
Parents: 0f5abf4
Author: Abhishek Agarwal <ab...@inmobi.com>
Authored: Tue Jan 26 23:29:50 2016 +0530
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Feb 5 14:40:15 2016 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/kafka/KafkaSpout.java | 8 +++----
.../apache/storm/kafka/PartitionManager.java | 2 +-
.../kafka/trident/TridentKafkaEmitter.java | 23 +++++++++++---------
3 files changed, 18 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5eac6855/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
index 7a83ae0..d1da446 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
@@ -17,17 +17,17 @@
*/
package org.apache.storm.kafka;
+import com.google.common.base.Strings;
+
import org.apache.storm.Config;
+import org.apache.storm.kafka.PartitionManager.KafkaMessageId;
import org.apache.storm.metric.api.IMetric;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
-import com.google.common.base.Strings;
-import kafka.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.storm.kafka.PartitionManager.KafkaMessageId;
import java.util.*;
@@ -100,7 +100,7 @@ public class KafkaSpout extends BaseRichSpout {
}
_kafkaOffsetMetric.refreshPartitions(latestPartitions);
for (PartitionManager pm : pms) {
- _kafkaOffsetMetric.setLatestEmittedOffset(pm.getPartition(), pm.lastCompletedOffset());
+ _kafkaOffsetMetric.setOffsetData(pm.getPartition(), pm.getOffsetData());
}
return _kafkaOffsetMetric.getValueAndReset();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/5eac6855/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
index bcc4001..dbf70a0 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/PartitionManager.java
@@ -320,7 +320,7 @@ public class PartitionManager {
}
}
- static class OffsetData {
+ public static class OffsetData {
public long latestEmittedOffset;
public long latestCompletedOffset;
http://git-wip-us.apache.org/repos/asf/storm/blob/5eac6855/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
index 6eddaf5..9732c8c 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
@@ -17,27 +17,28 @@
*/
package org.apache.storm.kafka.trident;
+import com.google.common.collect.ImmutableMap;
+
import org.apache.storm.Config;
+import org.apache.storm.kafka.*;
import org.apache.storm.metric.api.CombinedMetric;
import org.apache.storm.metric.api.MeanReducer;
import org.apache.storm.metric.api.ReducedMetric;
import org.apache.storm.task.TopologyContext;
-import com.google.common.collect.ImmutableMap;
-import kafka.javaapi.consumer.SimpleConsumer;
-import kafka.javaapi.message.ByteBufferMessageSet;
-import kafka.message.Message;
-import kafka.message.MessageAndOffset;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.storm.kafka.*;
-import org.apache.storm.kafka.TopicOffsetOutOfRangeException;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
import org.apache.storm.trident.spout.IPartitionedTridentSpout;
import org.apache.storm.trident.topology.TransactionAttempt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.*;
+import kafka.javaapi.consumer.SimpleConsumer;
+import kafka.javaapi.message.ByteBufferMessageSet;
+import kafka.message.Message;
+import kafka.message.MessageAndOffset;
+
public class TridentKafkaEmitter {
public static final Logger LOG = LoggerFactory.getLogger(TridentKafkaEmitter.class);
@@ -65,7 +66,9 @@ public class TridentKafkaEmitter {
private Map failFastEmitNewPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map lastMeta) {
SimpleConsumer consumer = _connections.register(partition);
Map ret = doEmitNewPartitionBatch(consumer, partition, collector, lastMeta);
- _kafkaOffsetMetric.setLatestEmittedOffset(partition, (Long) ret.get("offset"));
+ Long offset = (Long) ret.get("offset");
+ Long endOffset = (Long) ret.get("nextOffset");
+ _kafkaOffsetMetric.setOffsetData(partition, new PartitionManager.OffsetData(endOffset, offset));
return ret;
}
[3/3] storm git commit: add STORM-1455 to CHANGELOG.md
Posted by ka...@apache.org.
add STORM-1455 to CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/7090be45
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/7090be45
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/7090be45
Branch: refs/heads/1.x-branch
Commit: 7090be454dcbe24fba819a92b58af8b468bbe728
Parents: 5eac685
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Feb 5 14:48:16 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Feb 5 14:48:16 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/7090be45/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5745859..df1ddb6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.0.0
+ * STORM-1455: kafka spout should not reset to the beginning of partition when offsetoutofrange exception occurs
* STORM-1518: Backport of STORM-1504
* STORM-1505: Add map, flatMap and filter functions in trident stream
* STORM-1510: Fix broken nimbus log link