You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by hm...@apache.org on 2017/12/07 22:26:05 UTC
storm git commit: STORM-2835: storm-kafka-client KafkaSpout can fail
to remove all tuples from waitingToEmit
Repository: storm
Updated Branches:
refs/heads/master 00c247742 -> c8b306dc3
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/c8b306dc
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c8b306dc
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c8b306dc
Branch: refs/heads/master
Commit: c8b306dc33e606c209046374952b6c9f7cc69ca0
Parents: 00c2477
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 13:42:40 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/c8b306dc/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 425681d..9d12307 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
@@ -242,9 +242,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
}
- if (waitingToEmit()) {
- emit();
- }
+ emitIfWaitingNotEmitted();
} else {
LOG.debug("Spout not initialized. Not sending tuples until initialization completes");
}
@@ -266,7 +264,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();
}
@@ -297,7 +295,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
return pollablePartitions;
}
- private boolean waitingToEmit() {
+ private boolean isWaitingToEmit() {
return waitingToEmit != null && waitingToEmit.hasNext();
}
@@ -372,19 +370,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(tp, record.offset());