You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2018/04/03 20:50:52 UTC
storm git commit: STORM-2994: KafkaSpout consumes messages but
doesn't commit offsets
Repository: storm
Updated Branches:
refs/heads/1.1.x-branch f2cdcae34 -> 44b687a99
STORM-2994: KafkaSpout consumes messages but doesn't commit offsets
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/44b687a9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/44b687a9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/44b687a9
Branch: refs/heads/1.1.x-branch
Commit: 44b687a997c5ce0c26c103f019094361f3c38af1
Parents: f2cdcae
Author: Rui Abreu <ru...@gmail.com>
Authored: Tue Apr 3 10:21:05 2018 +0100
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Tue Apr 3 22:49:58 2018 +0200
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 32 +++---
.../storm/kafka/spout/KafkaSpoutMessageId.java | 28 ++---
.../kafka/spout/KafkaSpoutNullTupleTest.java | 103 +++++++++++++++++++
3 files changed, 137 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/44b687a9/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 4464e68..e7f5156 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
@@ -289,7 +289,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
if (isAtLeastOnceProcessing()) {
commitOffsetsForAckedTuples(kafkaConsumer.assignment());
} else if (kafkaSpoutConfig.getProcessingGuarantee() == ProcessingGuarantee.NO_GUARANTEE) {
- Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
createFetchedOffsetsMetadata(kafkaConsumer.assignment());
kafkaConsumer.commitAsync(offsetsToCommit, null);
LOG.debug("Committed offsets {} to Kafka", offsetsToCommit);
@@ -386,7 +386,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
numPolledRecords);
if (kafkaSpoutConfig.getProcessingGuarantee() == KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) {
//Commit polled records immediately to ensure delivery is at-most-once.
- Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
createFetchedOffsetsMetadata(kafkaConsumer.assignment());
kafkaConsumer.commitSync(offsetsToCommit);
LOG.debug("Committed offsets {} to Kafka", offsetsToCommit);
@@ -500,8 +500,11 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
return true;
}
} else {
+ /*if a null tuple is not configured to be emitted, it should be marked as emitted and acked immediately
+ * to allow its offset to be commited to Kafka*/
LOG.debug("Not emitting null tuple for record [{}] as defined in configuration.", record);
- msgId.setEmitted(false);
+ msgId.setNullTuple(true);
+ offsetManagers.get(tp).addToEmitMsgs(msgId.offset());
ack(msgId);
}
}
@@ -522,7 +525,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
return offsetsToCommit;
}
-
+
private void commitOffsetsForAckedTuples(Set<TopicPartition> assignedPartitions) {
// Find offsets that are ready to be committed for every assigned topic partition
final Map<TopicPartition, OffsetManager> assignedOffsetManagers = new HashMap<>();
@@ -593,17 +596,22 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
// Only need to keep track of acked tuples if commits to Kafka are controlled by
// tuple acks, which happens only for at-least-once processing semantics
final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId;
+
+ if (msgId.isNullTuple()) {
+ //a null tuple should be added to the ack list since by definition is a direct ack
+ offsetManagers.get(msgId.getTopicPartition()).addToAckMsgs(msgId);
+ LOG.debug("Received direct ack for message [{}], associated with null tuple", msgId);
+ tupleListener.onAck(msgId);
+ return;
+ }
+
if (!emitted.contains(msgId)) {
- if (msgId.isEmitted()) {
- LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that "
- + "came from a topic-partition that this consumer group instance is no longer tracking "
- + "due to rebalance/partition reassignment. No action taken.", msgId);
- } else {
- LOG.debug("Received direct ack for message [{}], associated with null tuple", msgId);
- }
+ LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that "
+ + "came from a topic-partition that this consumer group instance is no longer tracking "
+ + "due to rebalance/partition reassignment. No action taken.", msgId);
} else {
Validate.isTrue(!retryService.isScheduled(msgId), "The message id " + msgId + " is queued for retry while being acked."
- + " This should never occur barring errors in the RetryService implementation or the spout code.");
+ + " This should never occur barring errors in the RetryService implementation or the spout code.");
offsetManagers.get(msgId.getTopicPartition()).addToAckMsgs(msgId);
emitted.remove(msgId);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/44b687a9/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutMessageId.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutMessageId.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutMessageId.java
index 1626fee..ddf6391 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutMessageId.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutMessageId.java
@@ -27,33 +27,33 @@ public class KafkaSpoutMessageId implements Serializable {
private final long offset;
private int numFails = 0;
/**
- * true if the record was emitted using a form of collector.emit(...). false
+ * false if the record was emitted using a form of collector.emit(...). true
* when skipping null tuples as configured by the user in KafkaSpoutConfig
*/
- private boolean emitted;
+ private boolean nullTuple;
public KafkaSpoutMessageId(ConsumerRecord<?, ?> consumerRecord) {
- this(consumerRecord, true);
+ this(consumerRecord, false);
}
- public KafkaSpoutMessageId(ConsumerRecord<?, ?> consumerRecord, boolean emitted) {
- this(new TopicPartition(consumerRecord.topic(), consumerRecord.partition()), consumerRecord.offset(), emitted);
+ public KafkaSpoutMessageId(ConsumerRecord<?, ?> consumerRecord, boolean nullTuple) {
+ this(new TopicPartition(consumerRecord.topic(), consumerRecord.partition()), consumerRecord.offset(), nullTuple);
}
public KafkaSpoutMessageId(TopicPartition topicPart, long offset) {
- this(topicPart, offset, true);
+ this(topicPart, offset, false);
}
/**
* Creates a new KafkaSpoutMessageId.
* @param topicPart The topic partition this message belongs to
* @param offset The offset of this message
- * @param emitted True iff this message is not being skipped as a null tuple
+ * @param nullTuple True if this message is being skipped as a null tuple
*/
- public KafkaSpoutMessageId(TopicPartition topicPart, long offset, boolean emitted) {
+ public KafkaSpoutMessageId(TopicPartition topicPart, long offset, boolean nullTuple) {
this.topicPart = topicPart;
this.offset = offset;
- this.emitted = emitted;
+ this.nullTuple = nullTuple;
}
public int partition() {
@@ -80,12 +80,12 @@ public class KafkaSpoutMessageId implements Serializable {
return topicPart;
}
- public boolean isEmitted() {
- return emitted;
+ public boolean isNullTuple() {
+ return nullTuple;
}
- public void setEmitted(boolean emitted) {
- this.emitted = emitted;
+ public void setNullTuple(boolean nullTuple) {
+ this.nullTuple = nullTuple;
}
@Override
@@ -94,7 +94,7 @@ public class KafkaSpoutMessageId implements Serializable {
+ "topic-partition=" + topicPart
+ ", offset=" + offset
+ ", numFails=" + numFails
- + ", emitted=" + emitted
+ + ", nullTuple=" + nullTuple
+ '}';
}
http://git-wip-us.apache.org/repos/asf/storm/blob/44b687a9/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutNullTupleTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutNullTupleTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutNullTupleTest.java
new file mode 100644
index 0000000..159366b
--- /dev/null
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutNullTupleTest.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka.spout;
+
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Time;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyList;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+public class KafkaSpoutNullTupleTest extends KafkaSpoutAbstractTest {
+
+ public KafkaSpoutNullTupleTest() {
+ super(2_000);
+ }
+
+
+ @Override
+ KafkaSpoutConfig<String, String> createSpoutConfig() {
+
+ return KafkaSpoutConfig.builder("127.0.0.1:" + kafkaUnitRule.getKafkaUnit().getKafkaPort(),
+ Pattern.compile(SingleTopicKafkaSpoutConfiguration.TOPIC))
+ .setOffsetCommitPeriodMs(commitOffsetPeriodMs)
+ .setRecordTranslator(new NullRecordExtractor())
+ .build();
+ }
+
+ @Test
+ public void testShouldCommitAllMessagesIfNotSetToEmitNullTuples() throws Exception {
+ final int messageCount = 10;
+ prepareSpout(messageCount);
+
+ //All null tuples should be commited, meaning they were considered by to be emitted and acked
+ for(int i = 0; i < messageCount; i++) {
+ spout.nextTuple();
+ }
+
+ verify(collectorMock,never()).emit(
+ anyString(),
+ anyList(),
+ any());
+
+ Time.advanceTime(commitOffsetPeriodMs + KafkaSpout.TIMER_DELAY_MS);
+ //Commit offsets
+ spout.nextTuple();
+
+ verifyAllMessagesCommitted(messageCount);
+ }
+
+ private class NullRecordExtractor implements RecordTranslator {
+
+ @Override
+ public List<Object> apply(ConsumerRecord record) {
+ return null;
+
+ }
+
+ @Override
+ public Fields getFieldsFor(String stream) {
+ return new Fields("topic", "key", "value");
+ }
+
+ /**
+ * @return the list of streams that this will handle.
+ */
+ @Override
+ public List<String> streams() {
+ return null;
+ }
+
+ @Override
+ public Object apply(Object record) {
+ return null;
+ }
+ }
+
+
+}