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/12/29 00:29:48 UTC
[2/4] storm git commit: STORM-2104: Rewrite a few parts to be
compatible with Java 7
STORM-2104: Rewrite a few parts to be compatible with Java 7
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/3e225b48
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/3e225b48
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/3e225b48
Branch: refs/heads/1.x-branch
Commit: 3e225b480cb83e60309893a10b52cd0dfaa7f673
Parents: a92b7d0
Author: Stig Rohde D�ssing <sd...@it-minds.dk>
Authored: Thu Dec 8 19:34:07 2016 +0100
Committer: Stig Rohde D�ssing <sd...@it-minds.dk>
Committed: Tue Dec 20 19:10:45 2016 +0100
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 8 ++++++-
.../kafka/spout/KafkaSpoutRebalanceTest.java | 24 ++++++++++++++++----
.../builders/TopicKeyValueTupleBuilder.java | 1 -
3 files changed, 26 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/3e225b48/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 245ea1a..737c810 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
@@ -159,7 +159,13 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
//Emitted messages for partitions that are no longer assigned to this spout can't be acked, and they shouldn't be retried. Remove them from emitted.
Set<TopicPartition> partitionsSet = new HashSet(partitions);
- emitted.removeIf((msgId) -> !partitionsSet.contains(msgId.getTopicPartition()));
+ Iterator<KafkaSpoutMessageId> msgIdIterator = emitted.iterator();
+ while (msgIdIterator.hasNext()) {
+ KafkaSpoutMessageId msgId = msgIdIterator.next();
+ if (!partitionsSet.contains(msgId.getTopicPartition())) {
+ msgIdIterator.remove();
+ }
+ }
for (TopicPartition tp : partitions) {
final OffsetAndMetadata committedOffset = kafkaConsumer.committed(tp);
http://git-wip-us.apache.org/repos/asf/storm/blob/3e225b48/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
index 3e077ab..2d3eb56 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRebalanceTest.java
@@ -27,24 +27,33 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration;
+
import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.getKafkaSpoutStreams;
+
import org.apache.storm.kafka.spout.internal.KafkaConsumerFactory;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
+
import static org.mockito.Matchers.anyCollection;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.when;
+
import org.junit.Before;
import org.mockito.Captor;
+
import static org.mockito.Mockito.reset;
+
import org.mockito.MockitoAnnotations;
+
import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.getKafkaSpoutConfig;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.Matchers.hasKey;
import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.anyList;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -67,7 +76,12 @@ public class KafkaSpoutRebalanceTest {
collectorMock = mock(SpoutOutputCollector.class);
conf = new HashMap<>();
consumerMock = mock(KafkaConsumer.class);
- consumerFactoryMock = (kafkaSpoutConfig) -> consumerMock;
+ consumerFactoryMock = new KafkaConsumerFactory<String, String>(){
+ @Override
+ public KafkaConsumer<String, String> createConsumer(KafkaSpoutConfig<String, String> kafkaSpoutConfig) {
+ return consumerMock;
+ }
+ };
}
//Returns messageIds in order of emission
@@ -88,9 +102,9 @@ public class KafkaSpoutRebalanceTest {
//Make the consumer return a single message for each partition
Map<TopicPartition, List<ConsumerRecord<String, String>>> firstPartitionRecords = new HashMap<>();
- firstPartitionRecords.put(partitionThatWillBeRevoked, Collections.singletonList(new ConsumerRecord(partitionThatWillBeRevoked.topic(), partitionThatWillBeRevoked.partition(), 0L, "key", "value")));
+ firstPartitionRecords.put(partitionThatWillBeRevoked, Collections.singletonList(new ConsumerRecord<>(partitionThatWillBeRevoked.topic(), partitionThatWillBeRevoked.partition(), 0L, "key", "value")));
Map<TopicPartition, List<ConsumerRecord<String, String>>> secondPartitionRecords = new HashMap<>();
- secondPartitionRecords.put(assignedPartition, Collections.singletonList(new ConsumerRecord(assignedPartition.topic(), assignedPartition.partition(), 0L, "key", "value")));
+ secondPartitionRecords.put(assignedPartition, Collections.singletonList(new ConsumerRecord<>(assignedPartition.topic(), assignedPartition.partition(), 0L, "key", "value")));
when(consumerMock.poll(anyLong()))
.thenReturn(new ConsumerRecords(firstPartitionRecords))
.thenReturn(new ConsumerRecords(secondPartitionRecords))
@@ -99,11 +113,11 @@ public class KafkaSpoutRebalanceTest {
//Emit the messages
spout.nextTuple();
ArgumentCaptor<KafkaSpoutMessageId> messageIdForRevokedPartition = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collectorMock).emit(anyObject(), anyObject(), messageIdForRevokedPartition.capture());
+ verify(collectorMock).emit(anyString(), anyList(), messageIdForRevokedPartition.capture());
reset(collectorMock);
spout.nextTuple();
ArgumentCaptor<KafkaSpoutMessageId> messageIdForAssignedPartition = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collectorMock).emit(anyObject(), anyObject(), messageIdForAssignedPartition.capture());
+ verify(collectorMock).emit(anyString(), anyList(), messageIdForAssignedPartition.capture());
//Now rebalance
consumerRebalanceListener.onPartitionsRevoked(assignedPartitions);
http://git-wip-us.apache.org/repos/asf/storm/blob/3e225b48/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/TopicKeyValueTupleBuilder.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/TopicKeyValueTupleBuilder.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/TopicKeyValueTupleBuilder.java
index 492b79d..4f20b58 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/TopicKeyValueTupleBuilder.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/builders/TopicKeyValueTupleBuilder.java
@@ -1,4 +1,3 @@
-
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file