http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/DefaultRecordTranslatorTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/DefaultRecordTranslatorTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/DefaultRecordTranslatorTest.java new file mode 100644 index 0000000..681953d --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/DefaultRecordTranslatorTest.java @@ -0,0 +1,37 @@ +/* + * 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 static org.junit.Assert.assertEquals; + +import java.util.Arrays; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.storm.tuple.Fields; +import org.junit.Test; + +public class DefaultRecordTranslatorTest { + @Test + public void testBasic() { + DefaultRecordTranslator<String, String> trans = new DefaultRecordTranslator<>(); + assertEquals(Arrays.asList("default"), trans.streams()); + assertEquals(new Fields("topic", "partition", "offset", "key", "value"), trans.getFieldsFor("default")); + ConsumerRecord<String, String> cr = new ConsumerRecord<>("TOPIC", 100, 100, "THE KEY", "THE VALUE"); + assertEquals(Arrays.asList("TOPIC", 100, 100l, "THE KEY", "THE VALUE"), trans.apply(cr)); + } +}
http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java new file mode 100644 index 0000000..0467383 --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java @@ -0,0 +1,179 @@ +/* + * 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.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.apache.storm.kafka.KafkaUnitRule; +import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; +import org.apache.storm.kafka.spout.internal.KafkaConsumerFactory; +import org.apache.storm.kafka.spout.internal.KafkaConsumerFactoryDefault; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.Time; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.MockitoAnnotations; + +import java.util.HashMap; +import java.util.Map; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +public abstract class KafkaSpoutAbstractTest { + @Rule + public KafkaUnitRule kafkaUnitRule = new KafkaUnitRule(); + + final TopologyContext topologyContext = mock(TopologyContext.class); + final Map<String, Object> conf = new HashMap<>(); + final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class); + final long commitOffsetPeriodMs; + + KafkaConsumer<String, String> consumerSpy; + KafkaSpout<String, String> spout; + + @Captor + ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture; + private Time.SimulatedTime simulatedTime; + private KafkaSpoutConfig<String, String> spoutConfig; + + /** + * This constructor should be called by the subclass' default constructor with the desired value + * @param commitOffsetPeriodMs commit offset period to be used in commit and verification of messages committed + */ + protected KafkaSpoutAbstractTest(long commitOffsetPeriodMs) { + this.commitOffsetPeriodMs = commitOffsetPeriodMs; + } + + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + + spoutConfig = createSpoutConfig(); + + consumerSpy = createConsumerSpy(); + + spout = new KafkaSpout<>(spoutConfig, createConsumerFactory()); + + simulatedTime = new Time.SimulatedTime(); + } + + private KafkaConsumerFactory<String, String> createConsumerFactory() { + + return new KafkaConsumerFactory<String, String>() { + @Override + public KafkaConsumer<String, String> createConsumer(KafkaSpoutConfig<String, String> kafkaSpoutConfig) { + return consumerSpy; + } + + }; + } + + KafkaConsumer<String, String> createConsumerSpy() { + return spy(new KafkaConsumerFactoryDefault<String, String>().createConsumer(spoutConfig)); + } + + @After + public void tearDown() throws Exception { + simulatedTime.close(); + } + + abstract KafkaSpoutConfig<String, String> createSpoutConfig(); + + void prepareSpout(int messageCount) throws Exception { + SingleTopicKafkaUnitSetupHelper.populateTopicData(kafkaUnitRule.getKafkaUnit(), SingleTopicKafkaSpoutConfiguration.TOPIC, messageCount); + SingleTopicKafkaUnitSetupHelper.initializeSpout(spout, conf, topologyContext, collectorMock); + } + + /** + * Helper method to in sequence do: + * <li> + * <ul>spout.nexTuple()</ul> + * <ul>verify messageId</ul> + * <ul>spout.ack(msgId)</ul> + * <ul>reset(collector) to be able to reuse mock</ul> + * </li> + * + * @param offset offset of message to be verified + * @return {@link ArgumentCaptor} of the messageId verified + */ + ArgumentCaptor<Object> nextTuple_verifyEmitted_ack_resetCollector(int offset) { + spout.nextTuple(); + + ArgumentCaptor<Object> messageId = verifyMessageEmitted(offset); + + spout.ack(messageId.getValue()); + + reset(collectorMock); + + return messageId; + } + + // offset and messageId are used interchangeably + ArgumentCaptor<Object> verifyMessageEmitted(int offset) { + final ArgumentCaptor<Object> messageId = ArgumentCaptor.forClass(Object.class); + + verify(collectorMock).emit( + eq(SingleTopicKafkaSpoutConfiguration.STREAM), + eq(new Values(SingleTopicKafkaSpoutConfiguration.TOPIC, + Integer.toString(offset), + Integer.toString(offset))), + messageId.capture()); + + return messageId; + } + + void commitAndVerifyAllMessagesCommitted(long msgCount) { + // reset commit timer such that commit happens on next call to nextTuple() + Time.advanceTime(commitOffsetPeriodMs + KafkaSpout.TIMER_DELAY_MS); + + //Commit offsets + spout.nextTuple(); + + verifyAllMessagesCommitted(msgCount); + } + + /* + * Asserts that commitSync has been called once, + * that there are only commits on one topic, + * and that the committed offset covers messageCount messages + */ + void verifyAllMessagesCommitted(long messageCount) { + verify(consumerSpy).commitSync(commitCapture.capture()); + + final Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue(); + assertThat("Expected commits for only one topic partition", commits.entrySet().size(), is(1)); + + OffsetAndMetadata offset = commits.entrySet().iterator().next().getValue(); + assertThat("Expected committed offset to cover all emitted messages", offset.offset(), is(messageCount)); + + reset(consumerSpy); + } +} http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java new file mode 100644 index 0000000..90e906b --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java @@ -0,0 +1,241 @@ +/* + * 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 static org.hamcrest.CoreMatchers.instanceOf; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.nullValue; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy; +import org.hamcrest.CoreMatchers; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +public class KafkaSpoutConfigTest { + + @Rule + public ExpectedException expectedException = ExpectedException.none(); + + @Test + public void testBasic() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic").build(); + assertEquals(FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST, conf.getFirstPollOffsetStrategy()); + assertNull(conf.getConsumerGroupId()); + assertTrue(conf.getTranslator() instanceof DefaultRecordTranslator); + HashMap<String, Object> expected = new HashMap<>(); + expected.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234"); + expected.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + expected.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + expected.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + expected.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + assertEquals(expected, conf.getKafkaProps()); + assertEquals(KafkaSpoutConfig.DEFAULT_METRICS_TIME_BUCKET_SIZE_SECONDS, conf.getMetricsTimeBucketSizeInSecs()); + } + + @Test + public void testSetEmitNullTuplesToTrue() { + final KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setEmitNullTuples(true) + .build(); + + assertTrue("Failed to set emit null tuples to true", conf.isEmitNullTuples()); + } + + @Test + public void testShouldNotChangeAutoOffsetResetPolicyWhenNotUsingAtLeastOnce() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) + .build(); + + assertThat("When at-least-once is not specified, the spout should use the Kafka default auto offset reset policy", + conf.getKafkaProps().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), nullValue()); + } + + @Test + public void testWillRespectExplicitAutoOffsetResetPolicy() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none") + .build(); + + assertThat("Should allow users to pick a different auto offset reset policy than the one recommended for the at-least-once processing guarantee", + (String)conf.getKafkaProps().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), is("none")); + } + + @Test + public void testCanConfigureWithExplicitTrueBooleanAutoCommitMode() { + /* + * Since adding setProcessingGuarantee to KafkaSpoutConfig we don't want users to set "enable.auto.commit" in the consumer config, + * because setting the processing guarantee will do it automatically. For backward compatibility we need to be able to handle the + * property being set anyway for a few releases, and try to set a processing guarantee that corresponds to the property. + */ + + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true) + .build(); + + assertThat("When setting enable auto commit to true explicitly the spout should use the 'none' processing guarantee", + conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE)); + } + + @Test + public void testCanConfigureWithExplicitFalseBooleanAutoCommitMode() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false) + .build(); + + assertThat("When setting enable auto commit to false explicitly the spout should use the 'at-least-once' processing guarantee", + conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE)); + } + + @Test + public void testCanConfigureWithExplicitTrueStringAutoCommitMode() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true") + .build(); + + assertThat("When setting enable auto commit to true explicitly the spout should use the 'none' processing guarantee", + conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE)); + } + + @Test + public void testCanConfigureWithExplicitFalseStringAutoCommitMode() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false") + .build(); + + assertThat("When setting enable auto commit explicitly to false the spout should use the 'at-least-once' processing guarantee", + conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE)); + } + + @Test + public void testCanGetKeyDeserializerWhenUsingDefaultBuilder() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .build(); + + assertThat("When using the default builder methods, the key deserializer should default to StringDeserializer", + conf.getKeyDeserializer(), instanceOf(StringDeserializer.class)); + } + + @Test + public void testCanGetValueDeserializerWhenUsingDefaultBuilder() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .build(); + + assertThat("When using the default builder methods, the value deserializer should default to StringDeserializer", + conf.getValueDeserializer(), instanceOf(StringDeserializer.class)); + } + + @Test + public void testCanOverrideDeprecatedDeserializerClassWithKafkaProps() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setKey(StringDeserializer.class) + .setValue(StringDeserializer.class) + .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .build(); + + assertThat("The last set key deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class)); + assertThat("The last set value deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class)); + } + + private static class SerializableStringDeserializer implements SerializableDeserializer { + + private final StringDeserializer delegate = new StringDeserializer(); + + @Override + public void configure(Map configs, boolean isKey) { + delegate.configure(configs, isKey); + } + + @Override + public Object deserialize(String topic, byte[] data) { + return delegate.deserialize(topic, data); + } + + @Override + public void close() { + delegate.close(); + } + } + + @Test + public void testCanOverrideDeprecatedDeserializerInstanceWithKafkaProps() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setKey(new SerializableStringDeserializer()) + .setValue(new SerializableStringDeserializer()) + .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .build(); + + assertThat("The last set key deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class)); + assertThat("The last set value deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class)); + } + + @Test + public void testCanOverrideKafkaPropsWithDeprecatedDeserializerSetter() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setKey(new SerializableStringDeserializer()) + .setValue(new SerializableStringDeserializer()) + .build(); + + assertThat("The last set key deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class)); + assertThat("The last set value deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class)); + } + + @Test + public void testCanMixOldAndNewDeserializerSetter() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setKey(new SerializableStringDeserializer()) + .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class) + .setValue(new SerializableStringDeserializer()) + .build(); + + assertThat("The last set key deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class)); + assertThat("The last set value deserializer should be used, regardless of how it is set", + conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class)); + } + + @Test + public void testMetricsTimeBucketSizeInSecs() { + KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic") + .setMetricsTimeBucketSizeInSecs(100) + .build(); + + assertEquals(100, conf.getMetricsTimeBucketSizeInSecs()); + } +} http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/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 new file mode 100755 index 0000000..dbba04b --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutEmitTest.java @@ -0,0 +1,214 @@ +/* + * Copyright 2017 The Apache Software Foundation. + * + * Licensed 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 static org.mockito.Matchers.anyLong; +import static org.mockito.Matchers.anyObject; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.ArrayList; +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.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.TopicPartition; +import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; +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.anyList; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.never; + +import org.apache.storm.utils.Time; +import org.apache.storm.utils.Time.SimulatedTime; +import org.junit.Before; +import org.mockito.InOrder; + +import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder; +import static org.mockito.Matchers.eq; + +public class KafkaSpoutEmitTest { + + private final long offsetCommitPeriodMs = 2_000; + private final TopologyContext contextMock = mock(TopologyContext.class); + private final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class); + private final Map<String, Object> conf = new HashMap<>(); + private final TopicPartition partition = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 1); + private KafkaConsumer<String, String> consumerMock; + private KafkaSpoutConfig<String, String> spoutConfig; + + @Before + public void setUp() { + spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setOffsetCommitPeriodMs(offsetCommitPeriodMs) + .build(); + consumerMock = mock(KafkaConsumer.class); + } + + @Test + public void testNextTupleEmitsAtMostOneTuple() { + //The spout should emit at most one message per call to nextTuple + //This is necessary for Storm to be able to throttle the spout according to maxSpoutPending + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>(); + records.put(partition, SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 10)); + + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<>(records)); + + spout.nextTuple(); + + verify(collectorMock, times(1)).emit(anyString(), anyList(), anyObject()); + } + + @Test + public void testNextTupleEmitsFailedMessagesEvenWhenMaxUncommittedOffsetsIsExceeded() throws IOException { + //The spout must reemit failed messages waiting for retry even if it is not allowed to poll for new messages due to maxUncommittedOffsets being exceeded + + //Emit maxUncommittedOffsets messages, and fail all of them. Then ensure that the spout will retry them when the retry backoff has passed + try (SimulatedTime simulatedTime = new SimulatedTime()) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>(); + int numRecords = spoutConfig.getMaxUncommittedOffsets(); + //This is cheating a bit since maxPollRecords would normally spread this across multiple polls + records.put(partition, SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, numRecords)); + + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<>(records)); + + for (int i = 0; i < numRecords; i++) { + spout.nextTuple(); + } + + ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock, times(numRecords)).emit(anyString(), anyList(), messageIds.capture()); + + for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) { + spout.fail(messageId); + } + + reset(collectorMock); + + Time.advanceTime(50); + //No backoff for test retry service, just check that messages will retry immediately + for (int i = 0; i < numRecords; i++) { + spout.nextTuple(); + } + + ArgumentCaptor<KafkaSpoutMessageId> retryMessageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock, times(numRecords)).emit(anyString(), anyList(), retryMessageIds.capture()); + + //Verify that the poll started at the earliest retriable tuple offset + List<Long> failedOffsets = new ArrayList<>(); + for (KafkaSpoutMessageId msgId : messageIds.getAllValues()) { + failedOffsets.add(msgId.offset()); + } + InOrder inOrder = inOrder(consumerMock); + inOrder.verify(consumerMock).seek(partition, failedOffsets.get(0)); + inOrder.verify(consumerMock).poll(anyLong()); + } + } + + @Test + public void testSpoutWillSkipPartitionsAtTheMaxUncommittedOffsetsLimit() { + //This verifies that partitions can't prevent each other from retrying tuples due to the maxUncommittedOffsets limit. + try (SimulatedTime simulatedTime = new SimulatedTime()) { + TopicPartition partitionTwo = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 2); + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition, partitionTwo); + 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.<String, String>createRecords(partition, 0, spoutConfig.getMaxUncommittedOffsets())); + records.put(partitionTwo, SpoutWithMockedConsumerSetupHelper.<String, String>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 < numMessages; i++) { + spout.nextTuple(); + } + + ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock, times(numMessages)).emit(anyString(), anyList(), messageIds.capture()); + + //Now fail a tuple on partition one and verify that it is allowed to retry, because the failed tuple is below the maxUncommittedOffsets limit + KafkaSpoutMessageId failedMessageIdPartitionOne = null; + for (KafkaSpoutMessageId msgId : messageIds.getAllValues()) { + if (msgId.partition() == partition.partition()) { + failedMessageIdPartitionOne = msgId; + break; + } + } + + spout.fail(failedMessageIdPartitionOne); + + //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. + KafkaSpoutMessageId failedMessageIdPartitionTwo = null; + for (KafkaSpoutMessageId msgId: messageIds.getAllValues()) { + if (msgId.partition() == partitionTwo.partition()) { + if (failedMessageIdPartitionTwo != null) { + if (msgId.offset() >= failedMessageIdPartitionTwo.offset()) { + failedMessageIdPartitionTwo = msgId; + } + } else { + failedMessageIdPartitionTwo = msgId; + } + } + } + + spout.fail(failedMessageIdPartitionTwo); + + reset(collectorMock); + + Time.advanceTime(50); + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, failedMessageIdPartitionOne.offset(), 1)))); + + spout.nextTuple(); + + verify(collectorMock, times(1)).emit(anyString(), anyList(), anyObject()); + + InOrder inOrder = inOrder(consumerMock); + inOrder.verify(consumerMock).seek(partition, failedMessageIdPartitionOne.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)); + + reset(collectorMock); + + //Now also check that no more tuples are polled for, since both partitions are at their limits + spout.nextTuple(); + + verify(collectorMock, never()).emit(anyString(), anyList(), anyObject()); + } + } + +} http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutLogCompactionSupportTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutLogCompactionSupportTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutLogCompactionSupportTest.java new file mode 100644 index 0000000..09f7fc5 --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutLogCompactionSupportTest.java @@ -0,0 +1,223 @@ +/* + * Copyright 2017 The Apache Software Foundation. + * + * Licensed 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 static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.consumer.*; +import org.apache.kafka.common.TopicPartition; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.utils.Time; +import org.apache.storm.utils.Time.SimulatedTime; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.InOrder; +import org.mockito.MockitoAnnotations; + +import static org.hamcrest.CoreMatchers.is; + +import static org.hamcrest.Matchers.hasKey; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.anyList; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; + +public class KafkaSpoutLogCompactionSupportTest { + + private final long offsetCommitPeriodMs = 2_000; + private final TopologyContext contextMock = mock(TopologyContext.class); + private final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class); + private final Map<String, Object> conf = new HashMap<>(); + private final TopicPartition partition = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 1); + private KafkaConsumer<String, String> consumerMock; + private KafkaSpoutConfig<String, String> spoutConfig; + + @Captor + private ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture; + + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setOffsetCommitPeriodMs(offsetCommitPeriodMs) + .build(); + consumerMock = mock(KafkaConsumer.class); + } + + @Test + public void testCommitSuccessWithOffsetVoids() { + //Verify that the commit logic can handle offset voids due to log compaction + try (SimulatedTime simulatedTime = new SimulatedTime()) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>(); + List<ConsumerRecord<String, String>> recordsForPartition = new ArrayList<>(); + // Offsets emitted are 0,1,2,3,4,<void>,8,9 + recordsForPartition.addAll(SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 5)); + recordsForPartition.addAll(SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 8, 2)); + records.put(partition, recordsForPartition); + + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<>(records)); + + for (int i = 0; i < recordsForPartition.size(); i++) { + spout.nextTuple(); + } + + ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock, times(recordsForPartition.size())).emit(anyString(), anyList(), messageIds.capture()); + + for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) { + spout.ack(messageId); + } + + // Advance time and then trigger first call to kafka consumer commit; the commit must progress to offset 9 + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<String, String>(Collections.<TopicPartition, List<ConsumerRecord<String, String>>>emptyMap())); + spout.nextTuple(); + + InOrder inOrder = inOrder(consumerMock); + inOrder.verify(consumerMock).commitSync(commitCapture.capture()); + inOrder.verify(consumerMock).poll(anyLong()); + + //verify that Offset 10 was last committed offset, since this is the offset the spout should resume at + Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue(); + assertTrue(commits.containsKey(partition)); + assertEquals(10, commits.get(partition).offset()); + } + } + + @Test + public void testWillSkipRetriableTuplesIfOffsetsAreCompactedAway() { + /* + Verify that failed offsets will only retry if the corresponding message exists. + When log compaction is enabled in Kafka it is possible that a tuple can fail, + and then be impossible to retry because the message in Kafka has been deleted. + The spout needs to quietly ack such tuples to allow commits to progress past the deleted offset. + */ + try (SimulatedTime simulatedTime = new SimulatedTime()) { + TopicPartition partitionTwo = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 2); + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition, partitionTwo); + + List<KafkaSpoutMessageId> firstPartitionMsgIds = SpoutWithMockedConsumerSetupHelper + .pollAndEmit(spout, consumerMock, 3, collectorMock, partition, 0, 1, 2); + reset(collectorMock); + List<KafkaSpoutMessageId> secondPartitionMsgIds = SpoutWithMockedConsumerSetupHelper + .pollAndEmit(spout, consumerMock, 3, collectorMock, partitionTwo, 0, 1, 2); + reset(collectorMock); + + for(int i = 0; i < 3; i++) { + spout.fail(firstPartitionMsgIds.get(i)); + spout.fail(secondPartitionMsgIds.get(i)); + } + + Time.advanceTime(50); + + //The failed tuples are ready for retry. Make it appear like 0 and 1 on the first partition were compacted away. + //In this case the second partition acts as control to verify that we only skip past offsets that are no longer present. + Map<TopicPartition, int[]> retryOffsets = new HashMap<>(); + retryOffsets.put(partition, new int[] {2}); + retryOffsets.put(partitionTwo, new int[] {0, 1, 2}); + int expectedEmits = 4; //2 on first partition, 0-2 on second partition + List<KafkaSpoutMessageId> retryMessageIds = SpoutWithMockedConsumerSetupHelper.pollAndEmit(spout, consumerMock, expectedEmits, collectorMock, retryOffsets); + + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); + spout.nextTuple(); + + verify(consumerMock).commitSync(commitCapture.capture()); + Map<TopicPartition, OffsetAndMetadata> committed = commitCapture.getValue(); + assertThat(committed.keySet(), is(Collections.singleton(partition))); + assertThat("The first partition should have committed up to the first retriable tuple that is not missing", committed.get(partition).offset(), is(2L)); + + for(KafkaSpoutMessageId msgId : retryMessageIds) { + spout.ack(msgId); + } + + //The spout should now commit all the offsets, since all offsets are either acked or were missing when retrying + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); + spout.nextTuple(); + + verify(consumerMock, times(2)).commitSync(commitCapture.capture()); + committed = commitCapture.getValue(); + assertThat(committed, hasKey(partition)); + assertThat(committed, hasKey(partitionTwo)); + assertThat(committed.get(partition).offset(), is(3L)); + assertThat(committed.get(partitionTwo).offset(), is(3L)); + } + } + + @Test + public void testWillSkipRetriableTuplesIfOffsetsAreCompactedAwayWithoutAckingPendingTuples() { + //Demonstrate that the spout doesn't ack pending tuples when skipping compacted tuples. The pending tuples should be allowed to finish normally. + try (SimulatedTime simulatedTime = new SimulatedTime()) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + List<KafkaSpoutMessageId> firstPartitionMsgIds = SpoutWithMockedConsumerSetupHelper + .pollAndEmit(spout, consumerMock, 3, collectorMock, partition, 0, 1, 2); + reset(collectorMock); + + spout.fail(firstPartitionMsgIds.get(0)); + spout.fail(firstPartitionMsgIds.get(2)); + + Time.advanceTime(50); + + //The failed tuples are ready for retry. Make it appear like 0 and 1 were compacted away. + List<KafkaSpoutMessageId> retryMessageIds = SpoutWithMockedConsumerSetupHelper.pollAndEmit(spout, consumerMock, 1, collectorMock, partition, 2); + for(KafkaSpoutMessageId msgId : retryMessageIds) { + spout.ack(msgId); + } + + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); + spout.nextTuple(); + + verify(consumerMock).commitSync(commitCapture.capture()); + Map<TopicPartition, OffsetAndMetadata> committed = commitCapture.getValue(); + assertThat(committed.keySet(), is(Collections.singleton(partition))); + assertThat("The first partition should have committed the missing offset, but no further since the next tuple is pending", + committed.get(partition).offset(), is(1L)); + + spout.ack(firstPartitionMsgIds.get(1)); + + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); + spout.nextTuple(); + + verify(consumerMock, times(2)).commitSync(commitCapture.capture()); + committed = commitCapture.getValue(); + assertThat(committed.keySet(), is(Collections.singleton(partition))); + assertThat("The first partition should have committed all offsets", committed.get(partition).offset(), is(3L)); + } + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutMessagingGuaranteeTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutMessagingGuaranteeTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutMessagingGuaranteeTest.java new file mode 100644 index 0000000..082cc58 --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutMessagingGuaranteeTest.java @@ -0,0 +1,259 @@ +/* + * Copyright 2017 The Apache Software Foundation. + * + * Licensed 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 static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.not; +import static org.hamcrest.CoreMatchers.nullValue; +import static org.junit.Assert.assertThat; +import static org.mockito.Matchers.anyList; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Matchers.argThat; +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.isNull; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +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.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.consumer.OffsetCommitCallback; +import org.apache.kafka.common.TopicPartition; +import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; +import org.apache.storm.kafka.spout.internal.CommitMetadataManager; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.utils.Time; +import org.apache.storm.utils.Time.SimulatedTime; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatcher; +import org.mockito.Captor; +import org.mockito.InOrder; +import org.mockito.runners.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class KafkaSpoutMessagingGuaranteeTest { + + @Captor + private ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture; + + private final TopologyContext contextMock = mock(TopologyContext.class); + private final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class); + private final Map<String, Object> conf = new HashMap<>(); + private final TopicPartition partition = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 1); + private KafkaConsumer<String, String> consumerMock; + + @Before + public void setUp() { + consumerMock = mock(KafkaConsumer.class); + } + + @Test + public void testAtMostOnceModeCommitsBeforeEmit() throws Exception { + //At-most-once mode must commit tuples before they are emitted to the topology to ensure that a spout crash won't cause replays. + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) + .build(); + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 1)))); + + spout.nextTuple(); + + when(consumerMock.position(partition)).thenReturn(1L); + + //The spout should have emitted the tuple, and must have committed it before emit + InOrder inOrder = inOrder(consumerMock, collectorMock); + inOrder.verify(consumerMock).poll(anyLong()); + inOrder.verify(consumerMock).commitSync(commitCapture.capture()); + inOrder.verify(collectorMock).emit(eq(SingleTopicKafkaSpoutConfiguration.STREAM), anyList()); + + CommitMetadataManager metadataManager = new CommitMetadataManager(contextMock, KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE); + Map<TopicPartition, OffsetAndMetadata> committedOffsets = commitCapture.getValue(); + assertThat(committedOffsets.get(partition).offset(), is(0L)); + assertThat(committedOffsets.get(partition).metadata(), is(metadataManager.getCommitMetadata())); + } + + private void doTestModeDisregardsMaxUncommittedOffsets(KafkaSpoutConfig<String, String> spoutConfig) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + when(consumerMock.poll(anyLong())) + .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, spoutConfig.getMaxUncommittedOffsets())))) + .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, spoutConfig.getMaxUncommittedOffsets() - 1, spoutConfig.getMaxUncommittedOffsets())))); + + for (int i = 0; i < spoutConfig.getMaxUncommittedOffsets() * 2; i++) { + spout.nextTuple(); + } + + verify(consumerMock, times(2)).poll(anyLong()); + verify(collectorMock, times(spoutConfig.getMaxUncommittedOffsets() * 2)).emit(eq(SingleTopicKafkaSpoutConfiguration.STREAM), anyList()); + } + + @Test + public void testAtMostOnceModeDisregardsMaxUncommittedOffsets() throws Exception { + //The maxUncommittedOffsets limit should not be enforced, since it is only meaningful in at-least-once mode + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) + .build(); + doTestModeDisregardsMaxUncommittedOffsets(spoutConfig); + } + + @Test + public void testNoGuaranteeModeDisregardsMaxUncommittedOffsets() throws Exception { + //The maxUncommittedOffsets limit should not be enforced, since it is only meaningful in at-least-once mode + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE) + .build(); + doTestModeDisregardsMaxUncommittedOffsets(spoutConfig); + } + + private void doTestModeCannotReplayTuples(KafkaSpoutConfig<String, String> spoutConfig) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 1)))); + + spout.nextTuple(); + + ArgumentCaptor<KafkaSpoutMessageId> msgIdCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock).emit(eq(SingleTopicKafkaSpoutConfiguration.STREAM), anyList(), msgIdCaptor.capture()); + assertThat("Should have captured a message id", msgIdCaptor.getValue(), not(nullValue())); + + spout.fail(msgIdCaptor.getValue()); + + reset(consumerMock); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 1, 1)))); + + spout.nextTuple(); + + //The consumer should not be seeking to retry the failed tuple, it should just be continuing from the current position + verify(consumerMock, never()).seek(eq(partition), anyLong()); + } + + @Test + public void testAtMostOnceModeCannotReplayTuples() throws Exception { + //When tuple tracking is enabled, the spout must not replay tuples in at-most-once mode + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) + .setTupleTrackingEnforced(true) + .build(); + doTestModeCannotReplayTuples(spoutConfig); + } + + @Test + public void testNoGuaranteeModeCannotReplayTuples() throws Exception { + //When tuple tracking is enabled, the spout must not replay tuples in no guarantee mode + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE) + .setTupleTrackingEnforced(true) + .build(); + doTestModeCannotReplayTuples(spoutConfig); + } + + @Test + public void testAtMostOnceModeDoesNotCommitAckedTuples() throws Exception { + //When tuple tracking is enabled, the spout must not commit acked tuples in at-most-once mode because they were committed before being emitted + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) + .setTupleTrackingEnforced(true) + .build(); + try (SimulatedTime time = new SimulatedTime()) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 1)))); + + spout.nextTuple(); + reset(consumerMock); + + ArgumentCaptor<KafkaSpoutMessageId> msgIdCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock).emit(eq(SingleTopicKafkaSpoutConfiguration.STREAM), anyList(), msgIdCaptor.capture()); + assertThat("Should have captured a message id", msgIdCaptor.getValue(), not(nullValue())); + + spout.ack(msgIdCaptor.getValue()); + + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + spoutConfig.getOffsetsCommitPeriodMs()); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.<TopicPartition, List<ConsumerRecord<String, String>>>emptyMap())); + + spout.nextTuple(); + + verify(consumerMock, never()).commitSync(argThat(new ArgumentMatcher<Map<TopicPartition, OffsetAndMetadata>>() { + @Override + public boolean matches(Object arg) { + Map<TopicPartition, OffsetAndMetadata> castArg = (Map<TopicPartition, OffsetAndMetadata>) arg; + return !castArg.containsKey(partition); + } + })); + } + } + + @Test + public void testNoGuaranteeModeCommitsPolledTuples() throws Exception { + //When using the no guarantee mode, the spout must commit tuples periodically, regardless of whether they've been acked + KafkaSpoutConfig<String, String> spoutConfig = createKafkaSpoutConfigBuilder(mock(Subscription.class), -1) + .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE) + .setTupleTrackingEnforced(true) + .build(); + + try (SimulatedTime time = new SimulatedTime()) { + KafkaSpout<String, String> spout = SpoutWithMockedConsumerSetupHelper.setupSpout(spoutConfig, conf, contextMock, collectorMock, consumerMock, partition); + + when(consumerMock.poll(anyLong())).thenReturn(new ConsumerRecords<>(Collections.singletonMap(partition, + SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partition, 0, 1)))); + + spout.nextTuple(); + + when(consumerMock.position(partition)).thenReturn(1L); + + ArgumentCaptor<KafkaSpoutMessageId> msgIdCaptor = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + verify(collectorMock).emit(eq(SingleTopicKafkaSpoutConfiguration.STREAM), anyList(), msgIdCaptor.capture()); + assertThat("Should have captured a message id", msgIdCaptor.getValue(), not(nullValue())); + + Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + spoutConfig.getOffsetsCommitPeriodMs()); + + spout.nextTuple(); + + verify(consumerMock).commitAsync(commitCapture.capture(), isNull(OffsetCommitCallback.class)); + + CommitMetadataManager metadataManager = new CommitMetadataManager(contextMock, KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE); + Map<TopicPartition, OffsetAndMetadata> committedOffsets = commitCapture.getValue(); + assertThat(committedOffsets.get(partition).offset(), is(1L)); + assertThat(committedOffsets.get(partition).metadata(), is(metadataManager.getCommitMetadata())); + } + } + +} http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java new file mode 100644 index 0000000..c2c46b5 --- /dev/null +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutReactivationTest.java @@ -0,0 +1,145 @@ +/* + * Copyright 2017 The Apache Software Foundation. + * + * Licensed 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 static org.apache.storm.kafka.spout.KafkaSpout.TIMER_DELAY_MS; +import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST; +import static org.mockito.Matchers.any; +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.reset; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.HashMap; +import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerConfig; +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.KafkaUnitRule; +import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration; +import org.apache.storm.kafka.spout.internal.KafkaConsumerFactory; +import org.apache.storm.kafka.spout.internal.KafkaConsumerFactoryDefault; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.utils.Time; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.runners.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class KafkaSpoutReactivationTest { + + @Rule + public KafkaUnitRule kafkaUnitRule = new KafkaUnitRule(); + + @Captor + private ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture; + + private final TopologyContext topologyContext = mock(TopologyContext.class); + private final Map<String, Object> conf = new HashMap<>(); + private final SpoutOutputCollector collector = mock(SpoutOutputCollector.class); + private final long commitOffsetPeriodMs = 2_000; + private KafkaConsumer<String, String> consumerSpy; + private KafkaConsumer<String, String> postReactivationConsumerSpy; + private KafkaSpout<String, String> spout; + private final int maxPollRecords = 10; + + @Before + public void setUp() { + KafkaSpoutConfig<String, String> spoutConfig = + SingleTopicKafkaSpoutConfiguration.setCommonSpoutConfig( + KafkaSpoutConfig.builder("127.0.0.1:" + kafkaUnitRule.getKafkaUnit().getKafkaPort(), + SingleTopicKafkaSpoutConfiguration.TOPIC)) + .setFirstPollOffsetStrategy(UNCOMMITTED_EARLIEST) + .setOffsetCommitPeriodMs(commitOffsetPeriodMs) + .setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords) + .build(); + KafkaConsumerFactory<String, String> consumerFactory = new KafkaConsumerFactoryDefault<>(); + this.consumerSpy = spy(consumerFactory.createConsumer(spoutConfig)); + this.postReactivationConsumerSpy = spy(consumerFactory.createConsumer(spoutConfig)); + KafkaConsumerFactory<String, String> consumerFactoryMock = mock(KafkaConsumerFactory.class); + when(consumerFactoryMock.createConsumer(any(KafkaSpoutConfig.class))) + .thenReturn(consumerSpy) + .thenReturn(postReactivationConsumerSpy); + this.spout = new KafkaSpout<>(spoutConfig, consumerFactoryMock); + } + + private void prepareSpout(int messageCount) throws Exception { + SingleTopicKafkaUnitSetupHelper.populateTopicData(kafkaUnitRule.getKafkaUnit(), SingleTopicKafkaSpoutConfiguration.TOPIC, messageCount); + SingleTopicKafkaUnitSetupHelper.initializeSpout(spout, conf, topologyContext, collector); + } + + private KafkaSpoutMessageId emitOne() { + ArgumentCaptor<KafkaSpoutMessageId> messageId = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); + spout.nextTuple(); + verify(collector).emit(anyString(), anyList(), messageId.capture()); + reset(collector); + return messageId.getValue(); + } + + @Test + public void testSpoutMustHandleReactivationGracefully() throws Exception { + try (Time.SimulatedTime time = new Time.SimulatedTime()) { + int messageCount = maxPollRecords * 2; + prepareSpout(messageCount); + + //Emit and ack some tuples, ensure that some polled tuples remain cached in the spout by emitting less than maxPollRecords + int beforeReactivationEmits = maxPollRecords - 3; + for (int i = 0; i < beforeReactivationEmits - 1; i++) { + KafkaSpoutMessageId msgId = emitOne(); + spout.ack(msgId); + } + + KafkaSpoutMessageId ackAfterDeactivateMessageId = emitOne(); + + //Cycle spout activation + spout.deactivate(); + SingleTopicKafkaUnitSetupHelper.verifyAllMessagesCommitted(consumerSpy, commitCapture, beforeReactivationEmits - 1); + //Tuples may be acked/failed after the spout deactivates, so we have to be able to handle this too + spout.ack(ackAfterDeactivateMessageId); + spout.activate(); + + //Emit and ack the rest + for (int i = beforeReactivationEmits; i < messageCount; i++) { + KafkaSpoutMessageId msgId = emitOne(); + spout.ack(msgId); + } + + //Commit + Time.advanceTime(TIMER_DELAY_MS + commitOffsetPeriodMs); + spout.nextTuple(); + + //Verify that no more tuples are emitted and all tuples are committed + SingleTopicKafkaUnitSetupHelper.verifyAllMessagesCommitted(postReactivationConsumerSpy, commitCapture, messageCount); + + reset(collector); + spout.nextTuple(); + verify(collector, never()).emit(anyString(), anyList(), any(KafkaSpoutMessageId.class)); + } + + } + +} http://git-wip-us.apache.org/repos/asf/storm/blob/e16fa19f/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 a554a3b..29d2a22 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 @@ -15,11 +15,26 @@ */ package org.apache.storm.kafka.spout; +import static org.hamcrest.CoreMatchers.not; +import static org.hamcrest.Matchers.hasKey; +import static org.junit.Assert.assertThat; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +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.Set; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -27,45 +42,31 @@ 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.apache.storm.utils.Time; +import org.apache.storm.utils.Time.SimulatedTime; +import org.junit.Before; 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.Mockito; 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; +import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.createKafkaSpoutConfigBuilder; +import static org.mockito.Matchers.anyObject; +import static org.mockito.Matchers.eq; public class KafkaSpoutRebalanceTest { @Captor private ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture; + private final long offsetCommitPeriodMs = 2_000; + private final Map<String, Object> conf = new HashMap<>(); private TopologyContext contextMock; private SpoutOutputCollector collectorMock; - private Map conf; private KafkaConsumer<String, String> consumerMock; private KafkaConsumerFactory<String, String> consumerFactoryMock; @@ -74,7 +75,6 @@ public class KafkaSpoutRebalanceTest { MockitoAnnotations.initMocks(this); contextMock = mock(TopologyContext.class); collectorMock = mock(SpoutOutputCollector.class); - conf = new HashMap<>(); consumerMock = mock(KafkaConsumer.class); consumerFactoryMock = new KafkaConsumerFactory<String, String>(){ @Override @@ -85,44 +85,39 @@ public class KafkaSpoutRebalanceTest { } //Returns messageIds in order of emission - private List<KafkaSpoutMessageId> emitOneMessagePerPartitionThenRevokeOnePartition(KafkaSpout<String, String> spout, TopicPartition partitionThatWillBeRevoked, TopicPartition assignedPartition) { - //Setup spout with mock consumer so we can get at the rebalance listener + private List<KafkaSpoutMessageId> emitOneMessagePerPartitionThenRevokeOnePartition(KafkaSpout<String, String> spout, TopicPartition partitionThatWillBeRevoked, TopicPartition assignedPartition, ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture) { + //Setup spout with mock consumer so we can get at the rebalance listener spout.open(conf, contextMock, collectorMock); spout.activate(); - ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture = ArgumentCaptor.forClass(ConsumerRebalanceListener.class); - verify(consumerMock).subscribe(anyList(), rebalanceListenerCapture.capture()); - //Assign partitions to the spout ConsumerRebalanceListener consumerRebalanceListener = rebalanceListenerCapture.getValue(); - List<TopicPartition> assignedPartitions = new ArrayList<>(); + Set<TopicPartition> assignedPartitions = new HashSet<>(); assignedPartitions.add(partitionThatWillBeRevoked); assignedPartitions.add(assignedPartition); consumerRebalanceListener.onPartitionsAssigned(assignedPartitions); + when(consumerMock.assignment()).thenReturn(assignedPartitions); //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"))); - Map<TopicPartition, List<ConsumerRecord<String, String>>> secondPartitionRecords = new HashMap<>(); - 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)) - .thenReturn(new ConsumerRecords(Collections.emptyMap())); + .thenReturn(new ConsumerRecords<>(Collections.singletonMap(partitionThatWillBeRevoked, SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(partitionThatWillBeRevoked, 0, 1)))) + .thenReturn(new ConsumerRecords<>(Collections.singletonMap(assignedPartition, SpoutWithMockedConsumerSetupHelper.<String, String>createRecords(assignedPartition, 0, 1)))) + .thenReturn(new ConsumerRecords<>(new HashMap<TopicPartition, List<ConsumerRecord<String, String>>>())); //Emit the messages spout.nextTuple(); ArgumentCaptor<KafkaSpoutMessageId> messageIdForRevokedPartition = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); - verify(collectorMock).emit(anyString(), anyList(), messageIdForRevokedPartition.capture()); + verify(collectorMock).emit(Mockito.anyString(), Mockito.anyList(), messageIdForRevokedPartition.capture()); reset(collectorMock); spout.nextTuple(); ArgumentCaptor<KafkaSpoutMessageId> messageIdForAssignedPartition = ArgumentCaptor.forClass(KafkaSpoutMessageId.class); - verify(collectorMock).emit(anyString(), anyList(), messageIdForAssignedPartition.capture()); + verify(collectorMock).emit(Mockito.anyString(), Mockito.anyList(), messageIdForAssignedPartition.capture()); //Now rebalance consumerRebalanceListener.onPartitionsRevoked(assignedPartitions); consumerRebalanceListener.onPartitionsAssigned(Collections.singleton(assignedPartition)); - + when(consumerMock.assignment()).thenReturn(Collections.singleton(assignedPartition)); + List<KafkaSpoutMessageId> emittedMessageIds = new ArrayList<>(); emittedMessageIds.add(messageIdForRevokedPartition.getValue()); emittedMessageIds.add(messageIdForAssignedPartition.getValue()); @@ -132,49 +127,122 @@ public class KafkaSpoutRebalanceTest { @Test public void spoutMustIgnoreAcksForTuplesItIsNotAssignedAfterRebalance() throws Exception { //Acking tuples for partitions that are no longer assigned is useless since the spout will not be allowed to commit them - KafkaSpout<String, String> spout = new KafkaSpout<>(getKafkaSpoutConfig(getKafkaSpoutStreams(), -1, 10), consumerFactoryMock); - String topic = SingleTopicKafkaSpoutConfiguration.TOPIC; - TopicPartition partitionThatWillBeRevoked = new TopicPartition(topic, 1); - TopicPartition assignedPartition = new TopicPartition(topic, 2); - - //Emit a message on each partition and revoke the first partition - List<KafkaSpoutMessageId> emittedMessageIds = emitOneMessagePerPartitionThenRevokeOnePartition(spout, partitionThatWillBeRevoked, assignedPartition); - - //Ack both emitted tuples - spout.ack(emittedMessageIds.get(0)); - spout.ack(emittedMessageIds.get(1)); - - //Ensure the commit timer has expired - Thread.sleep(510); - - //Make the spout commit any acked tuples - spout.nextTuple(); - //Verify that it only committed the message on the assigned partition - verify(consumerMock).commitSync(commitCapture.capture()); - - Map<TopicPartition, OffsetAndMetadata> commitCaptureMap = commitCapture.getValue(); - assertThat(commitCaptureMap, hasKey(assignedPartition)); - assertThat(commitCaptureMap, not(hasKey(partitionThatWillBeRevoked))); + try (SimulatedTime simulatedTime = new SimulatedTime()) { + ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture = ArgumentCaptor.forClass(ConsumerRebalanceListener.class); + Subscription subscriptionMock = mock(Subscription.class); + doNothing() + .when(subscriptionMock) + .subscribe(any(KafkaConsumer.class), rebalanceListenerCapture.capture(), any(TopologyContext.class)); + KafkaSpout<String, String> spout = new KafkaSpout<>(createKafkaSpoutConfigBuilder(subscriptionMock, -1) + .setOffsetCommitPeriodMs(offsetCommitPeriodMs) + .build(), consumerFactoryMock); + String topic = SingleTopicKafkaSpoutConfiguration.TOPIC; + TopicPartition partitionThatWillBeRevoked = new TopicPartition(topic, 1); + TopicPartition assignedPartition = new TopicPartition(topic, 2); + + //Emit a message on each partition and revoke the first partition + List<KafkaSpoutMessageId> emittedMessageIds = emitOneMessagePerPartitionThenRevokeOnePartition( + spout, partitionThatWillBeRevoked, assignedPartition, rebalanceListenerCapture); + + //Ack both emitted tuples + spout.ack(emittedMessageIds.get(0)); + spout.ack(emittedMessageIds.get(1)); + + //Ensure the commit timer has expired + Time.advanceTime(offsetCommitPeriodMs + KafkaSpout.TIMER_DELAY_MS); + //Make the spout commit any acked tuples + spout.nextTuple(); + //Verify that it only committed the message on the assigned partition + verify(consumerMock, times(1)).commitSync(commitCapture.capture()); + + Map<TopicPartition, OffsetAndMetadata> commitCaptureMap = commitCapture.getValue(); + assertThat(commitCaptureMap, hasKey(assignedPartition)); + assertThat(commitCaptureMap, not(hasKey(partitionThatWillBeRevoked))); + } } - + @Test public void spoutMustIgnoreFailsForTuplesItIsNotAssignedAfterRebalance() throws Exception { //Failing tuples for partitions that are no longer assigned is useless since the spout will not be allowed to commit them if they later pass + ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture = ArgumentCaptor.forClass(ConsumerRebalanceListener.class); + Subscription subscriptionMock = mock(Subscription.class); + doNothing() + .when(subscriptionMock) + .subscribe(any(KafkaConsumer.class), rebalanceListenerCapture.capture(), any(TopologyContext.class)); KafkaSpoutRetryService retryServiceMock = mock(KafkaSpoutRetryService.class); - KafkaSpout<String, String> spout = new KafkaSpout<>(getKafkaSpoutConfig(getKafkaSpoutStreams(), -1, 10, retryServiceMock), consumerFactoryMock); + KafkaSpout<String, String> spout = new KafkaSpout<>(createKafkaSpoutConfigBuilder(subscriptionMock, -1) + .setOffsetCommitPeriodMs(10) + .setRetry(retryServiceMock) + .build(), consumerFactoryMock); String topic = SingleTopicKafkaSpoutConfiguration.TOPIC; TopicPartition partitionThatWillBeRevoked = new TopicPartition(topic, 1); TopicPartition assignedPartition = new TopicPartition(topic, 2); - + + when(retryServiceMock.getMessageId(Mockito.any(ConsumerRecord.class))) + .thenReturn(new KafkaSpoutMessageId(partitionThatWillBeRevoked, 0)) + .thenReturn(new KafkaSpoutMessageId(assignedPartition, 0)); + //Emit a message on each partition and revoke the first partition - List<KafkaSpoutMessageId> emittedMessageIds = emitOneMessagePerPartitionThenRevokeOnePartition(spout, partitionThatWillBeRevoked, assignedPartition); + List<KafkaSpoutMessageId> emittedMessageIds = emitOneMessagePerPartitionThenRevokeOnePartition( + spout, partitionThatWillBeRevoked, assignedPartition, rebalanceListenerCapture); + + //Check that only two message ids were generated + verify(retryServiceMock, times(2)).getMessageId(Mockito.any(ConsumerRecord.class)); //Fail both emitted tuples spout.fail(emittedMessageIds.get(0)); spout.fail(emittedMessageIds.get(1)); - + //Check that only the tuple on the currently assigned partition is retried verify(retryServiceMock, never()).schedule(emittedMessageIds.get(0)); verify(retryServiceMock).schedule(emittedMessageIds.get(1)); } + + @Test + public void testReassignPartitionSeeksForOnlyNewPartitions() { + /* + * When partitions are reassigned, the spout should seek with the first poll offset strategy for new partitions. + * Previously assigned partitions should be left alone, since the spout keeps the emitted and acked state for those. + */ + + ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture = ArgumentCaptor.forClass(ConsumerRebalanceListener.class); + Subscription subscriptionMock = mock(Subscription.class); + doNothing() + .when(subscriptionMock) + .subscribe(any(KafkaConsumer.class), rebalanceListenerCapture.capture(), any(TopologyContext.class)); + KafkaSpout<String, String> spout = new KafkaSpout<>(createKafkaSpoutConfigBuilder(subscriptionMock, -1) + .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST) + .build(), consumerFactoryMock); + String topic = SingleTopicKafkaSpoutConfiguration.TOPIC; + TopicPartition assignedPartition = new TopicPartition(topic, 1); + TopicPartition newPartition = new TopicPartition(topic, 2); + + //Setup spout with mock consumer so we can get at the rebalance listener + spout.open(conf, contextMock, collectorMock); + spout.activate(); + + //Assign partitions to the spout + ConsumerRebalanceListener consumerRebalanceListener = rebalanceListenerCapture.getValue(); + Set<TopicPartition> assignedPartitions = new HashSet<>(); + assignedPartitions.add(assignedPartition); + consumerRebalanceListener.onPartitionsAssigned(assignedPartitions); + reset(consumerMock); + + //Set up committed so it looks like some messages have been committed on each partition + long committedOffset = 500; + when(consumerMock.committed(assignedPartition)).thenReturn(new OffsetAndMetadata(committedOffset)); + when(consumerMock.committed(newPartition)).thenReturn(new OffsetAndMetadata(committedOffset)); + + //Now rebalance and add a new partition + consumerRebalanceListener.onPartitionsRevoked(assignedPartitions); + Set<TopicPartition> newAssignedPartitions = new HashSet<>(); + newAssignedPartitions.add(assignedPartition); + newAssignedPartitions.add(newPartition); + consumerRebalanceListener.onPartitionsAssigned(newAssignedPartitions); + + //This partition was previously assigned, so the consumer position shouldn't change + verify(consumerMock, never()).seek(eq(assignedPartition), anyLong()); + //This partition is new, and should start at the committed offset + verify(consumerMock).seek(newPartition, committedOffset); + } }
