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 2017/12/16 08:11:05 UTC
[1/2] storm git commit: STORM-2851: Fix
ConcurrentModificationException in doSeekRetriablePartitions
Repository: storm
Updated Branches:
refs/heads/master e7fdd67a6 -> 06b48fe04
STORM-2851: Fix ConcurrentModificationException in doSeekRetriablePartitions
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/8e42652e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/8e42652e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/8e42652e
Branch: refs/heads/master
Commit: 8e42652ef40aede81c05c6e8a83da11889f9f54b
Parents: e7fdd67
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Sun Dec 10 19:22:14 2017 +0100
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Sat Dec 16 01:04:51 2017 +0100
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 62 +++++++++-------
.../storm/kafka/spout/KafkaSpoutEmitTest.java | 29 +++++---
.../kafka/spout/MaxUncommittedOffsetTest.java | 76 ++++++++++++++++----
3 files changed, 118 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/8e42652e/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 9d12307..030735b 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
@@ -34,6 +34,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.apache.commons.lang.Validate;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -229,14 +230,14 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
kafkaSpoutConfig.getSubscription().refreshAssignment();
}
- if (commit()) {
+ if (shouldCommit()) {
commitOffsetsForAckedTuples();
}
- Set<TopicPartition> pollablePartitions = poll();
- if (!pollablePartitions.isEmpty()) {
+ PollablePartitionsInfo pollablePartitionsInfo = getPollablePartitionsInfo();
+ if (pollablePartitionsInfo.shouldPoll()) {
try {
- setWaitingToEmit(pollKafkaBroker(pollablePartitions));
+ setWaitingToEmit(pollKafkaBroker(pollablePartitionsInfo));
} catch (RetriableException e) {
LOG.error("Failed to poll from kafka.", e);
}
@@ -257,23 +258,24 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
throw new RuntimeException(new InterruptedException("Kafka consumer was interrupted"));
}
- private boolean commit() {
+ private boolean shouldCommit() {
return isAtLeastOnceProcessing() && commitTimer.isExpiredResetOnTrue(); // timer != null for non auto commit mode
}
-
- private Set<TopicPartition> poll() {
- final int maxUncommittedOffsets = kafkaSpoutConfig.getMaxUncommittedOffsets();
-
+
+ private PollablePartitionsInfo getPollablePartitionsInfo() {
if (isWaitingToEmit()) {
LOG.debug("Not polling. Tuples waiting to be emitted.");
- return Collections.emptySet();
+ return new PollablePartitionsInfo(Collections.emptySet(), Collections.emptyMap());
}
+
Set<TopicPartition> assignment = kafkaConsumer.assignment();
if (!isAtLeastOnceProcessing()) {
- return assignment;
+ return new PollablePartitionsInfo(assignment, Collections.emptyMap());
}
+
Map<TopicPartition, Long> earliestRetriableOffsets = retryService.earliestRetriableOffsets();
Set<TopicPartition> pollablePartitions = new HashSet<>();
+ final int maxUncommittedOffsets = kafkaSpoutConfig.getMaxUncommittedOffsets();
for (TopicPartition tp : assignment) {
OffsetManager offsetManager = offsetManagers.get(tp);
int numUncommittedOffsets = offsetManager.getNumUncommittedOffsets();
@@ -292,7 +294,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
}
}
- return pollablePartitions;
+ return new PollablePartitionsInfo(pollablePartitions, earliestRetriableOffsets);
}
private boolean isWaitingToEmit() {
@@ -308,14 +310,14 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
// ======== poll =========
- private ConsumerRecords<K, V> pollKafkaBroker(Set<TopicPartition> pollablePartitions) {
- final Map<TopicPartition, Long> retriableOffsets = doSeekRetriableTopicPartitions(pollablePartitions);
+ private ConsumerRecords<K, V> pollKafkaBroker(PollablePartitionsInfo pollablePartitionsInfo) {
+ doSeekRetriableTopicPartitions(pollablePartitionsInfo.pollableEarliestRetriableOffsets);
Set<TopicPartition> pausedPartitions = new HashSet<>(kafkaConsumer.assignment());
- pausedPartitions.removeIf(pollablePartitions::contains);
+ pausedPartitions.removeIf(pollablePartitionsInfo.pollablePartitions::contains);
try {
kafkaConsumer.pause(pausedPartitions);
final ConsumerRecords<K, V> consumerRecords = kafkaConsumer.poll(kafkaSpoutConfig.getPollTimeoutMs());
- ackRetriableOffsetsIfCompactedAway(retriableOffsets, consumerRecords);
+ ackRetriableOffsetsIfCompactedAway(pollablePartitionsInfo.pollableEarliestRetriableOffsets, consumerRecords);
final int numPolledRecords = consumerRecords.count();
LOG.debug("Polled [{}] records from Kafka.",
numPolledRecords);
@@ -329,18 +331,11 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
}
}
- private Map<TopicPartition, Long> doSeekRetriableTopicPartitions(Set<TopicPartition> pollablePartitions) {
- final Map<TopicPartition, Long> retriableTopicPartitions = retryService.earliestRetriableOffsets();
- for (TopicPartition tp : retriableTopicPartitions.keySet()) {
- if (!pollablePartitions.contains(tp)) {
- retriableTopicPartitions.remove(tp);
- }
- }
- for (Entry<TopicPartition, Long> retriableTopicPartitionAndOffset : retriableTopicPartitions.entrySet()) {
+ private void doSeekRetriableTopicPartitions(Map<TopicPartition, Long> pollableEarliestRetriableOffsets) {
+ for (Entry<TopicPartition, Long> retriableTopicPartitionAndOffset : pollableEarliestRetriableOffsets.entrySet()) {
//Seek directly to the earliest retriable message for each retriable topic partition
kafkaConsumer.seek(retriableTopicPartitionAndOffset.getKey(), retriableTopicPartitionAndOffset.getValue());
}
- return retriableTopicPartitions;
}
private void ackRetriableOffsetsIfCompactedAway(Map<TopicPartition, Long> earliestRetriableOffsets,
@@ -625,4 +620,21 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
private String getTopicsString() {
return kafkaSpoutConfig.getSubscription().getTopicsString();
}
+
+ private static class PollablePartitionsInfo {
+ private final Set<TopicPartition> pollablePartitions;
+ //The subset of earliest retriable offsets that are on pollable partitions
+ private final Map<TopicPartition, Long> pollableEarliestRetriableOffsets;
+
+ public PollablePartitionsInfo(Set<TopicPartition> pollablePartitions, Map<TopicPartition, Long> earliestRetriableOffsets) {
+ this.pollablePartitions = pollablePartitions;
+ this.pollableEarliestRetriableOffsets = earliestRetriableOffsets.entrySet().stream()
+ .filter(entry -> pollablePartitions.contains(entry.getKey()))
+ .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
+ }
+
+ public boolean shouldPoll() {
+ return !this.pollablePartitions.isEmpty();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/8e42652e/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java
index 91831e4..a29624a 100755
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java
@@ -26,11 +26,9 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -51,6 +49,7 @@ import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyObject;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
public class KafkaSpoutEmitTest {
@@ -143,37 +142,47 @@ public class KafkaSpoutEmitTest {
Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>();
//This is cheating a bit since maxPollRecords would normally spread this across multiple polls
records.put(partition, SpoutWithMockedConsumerSetupHelper.createRecords(partition, 0, spoutConfig.getMaxUncommittedOffsets()));
- records.put(partitionTwo, SpoutWithMockedConsumerSetupHelper.createRecords(partitionTwo, 0, spoutConfig.getMaxUncommittedOffsets()));
+ records.put(partitionTwo, SpoutWithMockedConsumerSetupHelper.createRecords(partitionTwo, 0, spoutConfig.getMaxUncommittedOffsets() + 1));
+ int numMessages = spoutConfig.getMaxUncommittedOffsets()*2 + 1;
when(consumerMock.poll(anyLong()))
.thenReturn(new ConsumerRecords<>(records));
- for (int i = 0; i < spoutConfig.getMaxUncommittedOffsets()*2; i++) {
+ for (int i = 0; i < numMessages; i++) {
spout.nextTuple();
}
ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collectorMock, times(spoutConfig.getMaxUncommittedOffsets()*2)).emit(anyString(), anyList(), messageIds.capture());
+ verify(collectorMock, times(numMessages)).emit(anyString(), anyList(), messageIds.capture());
- //Now fail a tuple on partition 0 and verify that it is allowed to retry
- //Partition 1 should be paused, since it is at the uncommitted offsets limit
- Optional<KafkaSpoutMessageId> failedMessageId = messageIds.getAllValues().stream()
+ //Now fail a tuple on partition one and verify that it is allowed to retry, because the failed tuple is below the maxUncommittedOffsets limit
+ Optional<KafkaSpoutMessageId> failedMessageIdPartitionOne = messageIds.getAllValues().stream()
.filter(messageId -> messageId.partition() == partition.partition())
.findAny();
- spout.fail(failedMessageId.get());
+ spout.fail(failedMessageIdPartitionOne.get());
+
+ //Also fail the last tuple from partition two. Since the failed tuple is beyond the maxUncommittedOffsets limit, it should not be retried until earlier messages are acked.
+ Optional<KafkaSpoutMessageId> failedMessagePartitionTwo = messageIds.getAllValues().stream()
+ .filter(messageId -> messageId.partition() == partitionTwo.partition())
+ .max((msgId, msgId2) -> (int)(msgId.offset() - msgId2.offset()));
+
+ spout.fail(failedMessagePartitionTwo.get());
reset(collectorMock);
Time.advanceTime(50);
when(consumerMock.poll(anyLong()))
- .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, SpoutWithMockedConsumerSetupHelper.createRecords(partition, failedMessageId.get().offset(), 1))));
+ .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, SpoutWithMockedConsumerSetupHelper.createRecords(partition, failedMessageIdPartitionOne.get().offset(), 1))));
spout.nextTuple();
verify(collectorMock, times(1)).emit(anyObject(), anyObject(), anyObject());
InOrder inOrder = inOrder(consumerMock);
+ inOrder.verify(consumerMock).seek(partition, failedMessageIdPartitionOne.get().offset());
+ //Should not seek on the paused partition
+ inOrder.verify(consumerMock, never()).seek(eq(partitionTwo), anyLong());
inOrder.verify(consumerMock).pause(Collections.singleton(partitionTwo));
inOrder.verify(consumerMock).poll(anyLong());
inOrder.verify(consumerMock).resume(Collections.singleton(partitionTwo));
http://git-wip-us.apache.org/repos/asf/storm/blob/8e42652e/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java
index 501f733..c76fc15 100755
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/MaxUncommittedOffsetTest.java
@@ -45,8 +45,9 @@ import org.mockito.MockitoAnnotations;
import static org.apache.storm.kafka.spout.config.builder.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder;
import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.ArgumentMatchers.anyObject;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
public class MaxUncommittedOffsetTest {
@@ -98,8 +99,8 @@ public class MaxUncommittedOffsetTest {
spout.nextTuple();
}
verify(collector, times(maxUncommittedOffsets)).emit(
- anyObject(),
- anyObject(),
+ any(),
+ any(),
messageIds.capture());
return messageIds;
}
@@ -124,9 +125,9 @@ public class MaxUncommittedOffsetTest {
spout.nextTuple();
}
verify(collector, times(maxUncommittedOffsets)).emit(
- anyObject(),
- anyObject(),
- anyObject());
+ any(),
+ any(),
+ any());
}
}
@@ -153,9 +154,9 @@ public class MaxUncommittedOffsetTest {
}
verify(collector, times(0)).emit(
- anyObject(),
- anyObject(),
- anyObject());
+ any(),
+ any(),
+ any());
}
}
@@ -179,6 +180,53 @@ public class MaxUncommittedOffsetTest {
reached if the tuple at the maxUncommittedOffsets limit is the earliest retriable tuple,
or if the spout is 1 tuple below the limit, and receives a full maxPollRecords tuples in the poll.
*/
+
+ try (Time.SimulatedTime simulatedTime = new Time.SimulatedTime()) {
+ //First check that maxUncommittedOffsets is respected when emitting from scratch
+ ArgumentCaptor<KafkaSpoutMessageId> messageIds = emitMaxUncommittedOffsetsMessagesAndCheckNoMoreAreEmitted(numMessages);
+ reset(collector);
+
+ //Fail only the last tuple
+ List<KafkaSpoutMessageId> messageIdList = messageIds.getAllValues();
+ KafkaSpoutMessageId failedMessageId = messageIdList.get(messageIdList.size() - 1);
+ spout.fail(failedMessageId);
+
+ //Offset 0 to maxUncommittedOffsets - 2 are pending, maxUncommittedOffsets - 1 is failed but not retriable
+ //The spout should not emit any more tuples.
+ spout.nextTuple();
+ verify(collector, never()).emit(
+ any(),
+ any(),
+ any());
+
+ //Allow the failed record to retry
+ Time.advanceTimeSecs(initialRetryDelaySecs);
+ for (int i = 0; i < maxPollRecords; i++) {
+ spout.nextTuple();
+ }
+ ArgumentCaptor<KafkaSpoutMessageId> secondRunMessageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
+ verify(collector, times(maxPollRecords)).emit(
+ any(),
+ any(),
+ secondRunMessageIds.capture());
+ reset(collector);
+ assertThat(secondRunMessageIds.getAllValues().get(0), is(failedMessageId));
+
+ //There should now be maxUncommittedOffsets + maxPollRecords emitted in all.
+ //Fail the last emitted tuple and verify that the spout won't retry it because it's above the emit limit.
+ spout.fail(secondRunMessageIds.getAllValues().get(secondRunMessageIds.getAllValues().size() - 1));
+ Time.advanceTimeSecs(initialRetryDelaySecs);
+ spout.nextTuple();
+ verify(collector, never()).emit(any(), any(), any());
+ }
+ }
+
+ @Test
+ public void testNextTupleWillAllowRetryForTuplesBelowEmitLimit() throws Exception {
+ /*
+ For each partition the spout is allowed to retry all tuples between the committed offset, and maxUncommittedOffsets ahead.
+ It must retry tuples within that limit, even if more tuples were emitted.
+ */
try (Time.SimulatedTime simulatedTime = new Time.SimulatedTime()) {
//First check that maxUncommittedOffsets is respected when emitting from scratch
ArgumentCaptor<KafkaSpoutMessageId> messageIds = emitMaxUncommittedOffsetsMessagesAndCheckNoMoreAreEmitted(numMessages);
@@ -186,17 +234,17 @@ public class MaxUncommittedOffsetTest {
failAllExceptTheFirstMessageThenCommit(messageIds);
- //Offset 0 is acked, 1 to maxUncommittedOffsets - 1 are failed but not retriable
+ //Offset 0 is committed, 1 to maxUncommittedOffsets - 1 are failed but not retriable
//The spout should now emit another maxPollRecords messages
- //This is allowed because the acked message brings the numUncommittedOffsets below the cap
+ //This is allowed because the committed message brings the numUncommittedOffsets below the cap
for (int i = 0; i < maxUncommittedOffsets; i++) {
spout.nextTuple();
}
ArgumentCaptor<KafkaSpoutMessageId> secondRunMessageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
verify(collector, times(maxPollRecords)).emit(
- anyObject(),
- anyObject(),
+ any(),
+ any(),
secondRunMessageIds.capture());
reset(collector);
@@ -213,7 +261,7 @@ public class MaxUncommittedOffsetTest {
//Advance time so the failed tuples become ready for retry, and check that the spout will emit retriable tuples
//for all the failed tuples that are within maxUncommittedOffsets tuples of the committed offset
//This means 1 to maxUncommitteddOffsets, but not maxUncommittedOffsets+1...maxUncommittedOffsets+maxPollRecords-1
- for(KafkaSpoutMessageId msgId : secondRunMessageIds.getAllValues()) {
+ for (KafkaSpoutMessageId msgId : secondRunMessageIds.getAllValues()) {
spout.fail(msgId);
}
Time.advanceTimeSecs(initialRetryDelaySecs);
[2/2] storm git commit: Merge branch 'STORM-2851' of
https://github.com/srdo/storm into asfgit-master
Posted by sr...@apache.org.
Merge branch 'STORM-2851' of https://github.com/srdo/storm into asfgit-master
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/06b48fe0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/06b48fe0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/06b48fe0
Branch: refs/heads/master
Commit: 06b48fe0454d756c81dbedf906b5ff084a00c233
Parents: e7fdd67 8e42652
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Sat Dec 16 09:10:01 2017 +0100
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Sat Dec 16 09:10:01 2017 +0100
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 62 +++++++++-------
.../storm/kafka/spout/KafkaSpoutEmitTest.java | 29 +++++---
.../kafka/spout/MaxUncommittedOffsetTest.java | 76 ++++++++++++++++----
3 files changed, 118 insertions(+), 49 deletions(-)
----------------------------------------------------------------------