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 2017/10/09 23:14:24 UTC
[2/4] storm git commit: Backport to Java 7 and revert changes to
RetryService API
Backport to Java 7 and revert changes to RetryService API
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/7d290aff
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/7d290aff
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/7d290aff
Branch: refs/heads/1.1.x-branch
Commit: 7d290affa44265c49665331a83271c0c33051c93
Parents: 4971543
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Wed Oct 4 20:46:46 2017 +0200
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Wed Oct 4 21:09:00 2017 +0200
----------------------------------------------------------------------
.../apache/storm/kafka/spout/KafkaSpout.java | 10 ++++--
.../KafkaSpoutRetryExponentialBackoff.java | 5 +--
.../kafka/spout/KafkaSpoutRetryService.java | 13 ++++----
.../kafka/spout/KafkaSpoutRebalanceTest.java | 4 +--
.../KafkaSpoutRetryExponentialBackoffTest.java | 33 +++++++++++---------
.../kafka/spout/SingleTopicKafkaSpoutTest.java | 4 +--
6 files changed, 41 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/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 a667f8b..e0fde9c 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
@@ -156,7 +156,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 should not be retried, hence remove them from emitted collection.
*/
- emitted.removeIf(msgId -> !partitions.contains(msgId.getTopicPartition()));
+ Iterator<KafkaSpoutMessageId> msgIdIterator = emitted.iterator();
+ while (msgIdIterator.hasNext()) {
+ KafkaSpoutMessageId msgId = msgIdIterator.next();
+ if (!partitions.contains(msgId.getTopicPartition())) {
+ msgIdIterator.remove();
+ }
+ }
}
Set<TopicPartition> newPartitions = new HashSet<>(partitions);
@@ -312,7 +318,7 @@ public class KafkaSpout<K, V> extends BaseRichSpout {
*/
private boolean emitTupleIfNotEmitted(ConsumerRecord<K, V> record) {
final TopicPartition tp = new TopicPartition(record.topic(), record.partition());
- final KafkaSpoutMessageId msgId = retryService.getMessageId(tp, record.offset());
+ final KafkaSpoutMessageId msgId = retryService.getMessageId(record);
if (offsetManagers.containsKey(tp) && offsetManagers.get(tp).contains(msgId)) { // has been acked
LOG.trace("Tuple for record [{}] has already been acked. Skipping", record);
} else if (emitted.contains(msgId)) { // has been emitted and it's pending ack or fail
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
index 81ef986..68a6f3f 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
@@ -33,6 +33,7 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.Validate;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.storm.utils.Time;
/**
@@ -289,8 +290,8 @@ public class KafkaSpoutRetryExponentialBackoff implements KafkaSpoutRetryService
}
@Override
- public KafkaSpoutMessageId getMessageId(TopicPartition tp, long offset) {
- KafkaSpoutMessageId msgId = new KafkaSpoutMessageId(tp, offset);
+ public KafkaSpoutMessageId getMessageId(ConsumerRecord<?, ?> record) {
+ KafkaSpoutMessageId msgId = new KafkaSpoutMessageId(record);
if (isScheduled(msgId)) {
for (KafkaSpoutMessageId originalMsgId : toRetryMsgs) {
if (originalMsgId.equals(msgId)) {
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryService.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryService.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryService.java
index b70acd7..12d26da 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryService.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryService.java
@@ -18,11 +18,11 @@
package org.apache.storm.kafka.spout;
-import org.apache.kafka.common.TopicPartition;
import java.io.Serializable;
import java.util.Collection;
import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
/**
@@ -84,10 +84,11 @@ public interface KafkaSpoutRetryService extends Serializable {
int readyMessageCount();
/**
- * Gets the {@link KafkaSpoutMessageId} for the record on the given topic partition and offset.
- * @param topicPartition The topic partition of the record
- * @param offset The offset of the record
- * @return The id the record was scheduled for retry with, or a new {@link KafkaSpoutMessageId} if the record was not scheduled for retry.
+ * Gets the {@link KafkaSpoutMessageId} for the given record.
+ *
+ * @param record The record to fetch the id for
+ * @return The id the record was scheduled for retry with, or a new {@link KafkaSpoutMessageId} if the record was not scheduled for
+ * retry.
*/
- KafkaSpoutMessageId getMessageId(TopicPartition topicPartition, long offset);
+ KafkaSpoutMessageId getMessageId(ConsumerRecord<?, ?> record);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/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 988dbb8..79a1c9a 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
@@ -197,10 +197,10 @@ public class KafkaSpoutRebalanceTest {
Subscription subscriptionMock = mock(Subscription.class);
doNothing()
.when(subscriptionMock)
- .subscribe(any(), rebalanceListenerCapture.capture(), any());
+ .subscribe(any(KafkaConsumer.class), rebalanceListenerCapture.capture(), any(TopologyContext.class));
KafkaSpout<String, String> spout = new KafkaSpout<>(createKafkaSpoutConfigBuilder(subscriptionMock, -1)
.setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)
- .build(), consumerFactory);
+ .build(), consumerFactoryMock);
String topic = SingleTopicKafkaSpoutConfiguration.TOPIC;
TopicPartition assignedPartition = new TopicPartition(topic, 1);
TopicPartition newPartition = new TopicPartition(topic, 2);
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoffTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoffTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoffTest.java
index c543f8b..f6de6a8 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoffTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoffTest.java
@@ -25,6 +25,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
import org.apache.storm.utils.Time;
@@ -44,17 +45,21 @@ public class KafkaSpoutRetryExponentialBackoffTest {
return new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(1), TimeInterval.seconds(0), 1, TimeInterval.seconds(1));
}
+ private ConsumerRecord<String, String> createRecord(TopicPartition tp, long offset) {
+ return new ConsumerRecord<>(tp.topic(), tp.partition(), offset, null, null);
+ }
+
@Test
public void testCanScheduleRetry() {
KafkaSpoutRetryExponentialBackoff retryService = createNoWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
boolean scheduled = retryService.schedule(msgId);
assertThat("The service must schedule the message for retry", scheduled, is(true));
- KafkaSpoutMessageId retrievedMessageId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId retrievedMessageId = retryService.getMessageId(createRecord(testTopic, offset));
assertThat("The service should return the original message id when asked for the same tp/offset twice", retrievedMessageId, sameInstance(msgId));
assertThat(retryService.isScheduled(msgId), is(true));
assertThat(retryService.isReady(msgId), is(true));
@@ -68,7 +73,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = createOneSecondWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
retryService.schedule(msgId);
@@ -79,7 +84,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
Time.advanceTime(500);
assertThat("The message should not be ready for retry yet since it was rescheduled", retryService.isReady(msgId), is(false));
assertThat(retryService.isScheduled(msgId), is(true));
- assertThat(retryService.earliestRetriableOffsets(), is(Collections.emptyMap()));
+ assertThat(retryService.earliestRetriableOffsets(), is(Collections.<TopicPartition, Long>emptyMap()));
assertThat(retryService.readyMessageCount(), is(0));
Time.advanceTime(500);
assertThat("The message should be ready for retry once the full delay has passed", retryService.isReady(msgId), is(true));
@@ -95,7 +100,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = createOneSecondWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
retryService.schedule(msgId);
@@ -113,7 +118,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
public void testCanRemoveRetry() {
KafkaSpoutRetryExponentialBackoff retryService = createNoWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
retryService.schedule(msgId);
@@ -122,7 +127,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
assertThat(removed, is(true));
assertThat(retryService.isScheduled(msgId), is(false));
assertThat(retryService.isReady(msgId), is(false));
- assertThat(retryService.earliestRetriableOffsets(), is(Collections.emptyMap()));
+ assertThat(retryService.earliestRetriableOffsets(), is(Collections.<TopicPartition, Long>emptyMap()));
assertThat(retryService.readyMessageCount(), is(0));
}
@@ -133,8 +138,8 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = createOneSecondWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgIdTp1 = retryService.getMessageId(testTopic, offset);
- KafkaSpoutMessageId msgIdTp2 = retryService.getMessageId(testTopic2, offset);
+ KafkaSpoutMessageId msgIdTp1 = retryService.getMessageId(createRecord(testTopic, offset));
+ KafkaSpoutMessageId msgIdTp2 = retryService.getMessageId(createRecord(testTopic2, offset));
msgIdTp1.incrementNumFails();
msgIdTp2.incrementNumFails();
@@ -179,8 +184,8 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = createOneSecondWaitRetryService();
long offset = 0;
- KafkaSpoutMessageId msgIdEarliest = retryService.getMessageId(testTopic, offset);
- KafkaSpoutMessageId msgIdLatest = retryService.getMessageId(testTopic, offset + 1);
+ KafkaSpoutMessageId msgIdEarliest = retryService.getMessageId(createRecord(testTopic, offset));
+ KafkaSpoutMessageId msgIdLatest = retryService.getMessageId(createRecord(testTopic, offset + 1));
msgIdEarliest.incrementNumFails();
msgIdLatest.incrementNumFails();
@@ -213,7 +218,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0), TimeInterval.seconds(0), maxRetries, TimeInterval.seconds(0));
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
for (int i = 0; i < maxRetries; i++) {
msgId.incrementNumFails();
}
@@ -240,7 +245,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(500), TimeInterval.seconds(0), 1, TimeInterval.seconds(maxDelaySecs));
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
retryService.schedule(msgId);
@@ -263,7 +268,7 @@ public class KafkaSpoutRetryExponentialBackoffTest {
KafkaSpoutRetryExponentialBackoff retryService = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0), TimeInterval.seconds(4), Integer.MAX_VALUE, TimeInterval.seconds(Integer.MAX_VALUE));
long offset = 0;
- KafkaSpoutMessageId msgId = retryService.getMessageId(testTopic, offset);
+ KafkaSpoutMessageId msgId = retryService.getMessageId(createRecord(testTopic, offset));
msgId.incrementNumFails();
msgId.incrementNumFails(); //First failure is the initial delay, so not interesting
http://git-wip-us.apache.org/repos/asf/storm/blob/7d290aff/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java
index ec31015..8ff6142 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/SingleTopicKafkaSpoutTest.java
@@ -141,7 +141,7 @@ public class SingleTopicKafkaSpoutTest {
spout.nextTuple();
}
ArgumentCaptor<KafkaSpoutMessageId> messageIdCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collector, times(messageCount)).emit(anyObject(), anyObject(), messageIdCaptor.capture());
+ verify(collector, times(messageCount)).emit(anyString(), anyList(), messageIdCaptor.capture());
List<KafkaSpoutMessageId> messageIds = messageIdCaptor.getAllValues();
for (int i = 1; i < messageIds.size(); i++) {
spout.ack(messageIds.get(i));
@@ -153,7 +153,7 @@ public class SingleTopicKafkaSpoutTest {
reset(collector);
spout.nextTuple();
ArgumentCaptor<KafkaSpoutMessageId> failedIdReplayCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collector).emit(anyObject(), anyObject(), failedIdReplayCaptor.capture());
+ verify(collector).emit(anyString(), anyList(), failedIdReplayCaptor.capture());
assertThat("Expected replay of failed tuple", failedIdReplayCaptor.getValue(), is(failedTuple));