You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2017/12/20 20:29:16 UTC
[09/23] storm git commit: STORM-2835: storm-kafka-client KafkaSpout
can fail to remove all tuples from waitingToEmit
STORM-2835: storm-kafka-client KafkaSpout can fail to remove all tuples from waitingToEmit
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/53c2c677
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/53c2c677
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/53c2c677
Branch: refs/heads/metrics_v2
Commit: 53c2c677e1066df1ac97b7a62a2d41cab484ecbd
Parents: 31133fd
Author: Hugo Louro <hm...@gmail.com>
Authored: Tue Nov 28 18:25:19 2017 -0800
Committer: Hugo Louro <hm...@gmail.com>
Committed: Thu Dec 7 14:03:33 2017 -0800
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 20 +++++++++++---------
1 file changed, 11 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/53c2c677/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
index b126770..36465f8 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
@@ -248,9 +248,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
}
- if (waitingToEmit()) {
- emit();
- }
+ emitIfWaitingNotEmitted();
} else {
LOG.debug("Spout not initialized. Not sending tuples until initialization completes");
}
@@ -272,7 +270,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
private Set<TopicPartition> poll() {
final int maxUncommittedOffsets = kafkaSpoutConfig.getMaxUncommittedOffsets();
- if (waitingToEmit()) {
+ if (isWaitingToEmit()) {
LOG.debug("Not polling. Tuples waiting to be emitted.");
return Collections.emptySet();
}
@@ -303,7 +301,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
return pollablePartitions;
}
- private boolean waitingToEmit() {
+ private boolean isWaitingToEmit() {
return waitingToEmit != null && waitingToEmit.hasNext();
}
@@ -383,19 +381,23 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
// ======== emit =========
- private void emit() {
- while (!emitTupleIfNotEmitted(waitingToEmit.next()) && waitingToEmit.hasNext()) {
+ private void emitIfWaitingNotEmitted() {
+ while (isWaitingToEmit()) {
+ final boolean emitted = emitOrRetryTuple(waitingToEmit.next());
waitingToEmit.remove();
+ if (emitted) {
+ break;
+ }
}
}
/**
- * Creates a tuple from the kafka record and emits it if it was not yet emitted.
+ * Creates a tuple from the kafka record and emits it if it was never emitted or it is ready to be retried.
*
* @param record to be emitted
* @return true if tuple was emitted. False if tuple has been acked or has been emitted and is pending ack or fail
*/
- private boolean emitTupleIfNotEmitted(ConsumerRecord<K, V> record) {
+ private boolean emitOrRetryTuple(ConsumerRecord<K, V> record) {
final TopicPartition tp = new TopicPartition(record.topic(), record.partition());
final KafkaSpoutMessageId msgId = retryService.getMessageId(record);