Repository: kafka Updated Branches: refs/heads/0.9.0 7684f6dd8 -> 416210ec5
KAFKA-2893: Add a simple non-negative partition seek check Author: jinxing <[email protected]> Author: ZoneMayor <[email protected]> Reviewers: Guozhang Wang Closes #628 from ZoneMayor/trunk-KAFKA-2893 (cherry picked from commit 9d23b512ccb95940782f51c93bd6cdbed3e938db) Signed-off-by: Guozhang Wang <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/416210ec Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/416210ec Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/416210ec Branch: refs/heads/0.9.0 Commit: 416210ec55fa0ca3dbda2b875feeb2db6ac38cc7 Parents: 7684f6d Author: Jin Xing <[email protected]> Authored: Thu Dec 10 10:49:02 2015 -0800 Committer: Guozhang Wang <[email protected]> Committed: Thu Dec 10 10:49:10 2015 -0800 ---------------------------------------------------------------------- .../apache/kafka/clients/consumer/KafkaConsumer.java | 3 +++ .../kafka/clients/consumer/KafkaConsumerTest.java | 14 ++++++++++++++ 2 files changed, 17 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/416210ec/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index f7397ec..70a3ae6 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1008,6 +1008,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> { */ @Override public void seek(TopicPartition partition, long offset) { + if (offset < 0) { + throw new IllegalArgumentException("seek offset must not be a negative number"); + } acquire(); try { log.debug("Seeking to offset {} for partition {}", offset, partition); http://git-wip-us.apache.org/repos/asf/kafka/blob/416210ec/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 983c45d..5711852 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -23,6 +23,7 @@ import org.apache.kafka.test.MockMetricsReporter; import org.junit.Assert; import org.junit.Test; +import java.util.Arrays; import java.util.Collections; import java.util.Properties; @@ -78,4 +79,17 @@ public class KafkaConsumerTest { Assert.assertTrue(consumer.subscription().isEmpty()); Assert.assertTrue(consumer.assignment().isEmpty()); } + + @Test(expected = IllegalArgumentException.class) + public void testSeekNegative() { + Properties props = new Properties(); + props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "testSeekNegative"); + props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + props.setProperty(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName()); + + KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>( + props, new ByteArrayDeserializer(), new ByteArrayDeserializer()); + consumer.assign(Arrays.asList(new TopicPartition("nonExistTopic", 0))); + consumer.seek(new TopicPartition("nonExistTopic", 0), -1); + } }
