From c032abd2bb6be97ca07b3827e150ddf2fabefa91 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 5 Aug 2015 17:05:58 -0700 Subject: [PATCH 01/18] KAFKA-2390; Seek() should take a callback --- .../kafka/clients/consumer/Consumer.java | 5 +++ .../consumer/ConsumerSeekCallback.java | 33 +++++++++++++++++++ .../kafka/clients/consumer/KafkaConsumer.java | 24 +++++++++++++- .../kafka/clients/consumer/MockConsumer.java | 7 +++- .../clients/consumer/internals/Fetcher.java | 3 +- .../consumer/internals/SubscriptionState.java | 24 ++++++++++++-- 6 files changed, 91 insertions(+), 5 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java index 158e1ea1e35db..1e052915afcea 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java @@ -83,6 +83,11 @@ public interface Consumer extends Closeable { */ public void seek(TopicPartition partition, long offset); + /** + * @see KafkaConsumer#seek(TopicPartition, long, ConsumerSeekCallback) + */ + public void seek(TopicPartition partition, long offset, ConsumerSeekCallback callbak); + /** * @see KafkaConsumer#seekToBeginning(TopicPartition...) */ diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java new file mode 100644 index 0000000000000..b4c619fbaba10 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java @@ -0,0 +1,33 @@ +/** + * 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.kafka.clients.consumer; + +import org.apache.kafka.common.TopicPartition; + +import java.util.Map; + +/** + * A callback interface that the user can implement to trigger custom actions when a fetch request after a seek completes. The callback + * may be executed in any thread calling {@link Consumer#poll(long) poll()}. + */ +public interface ConsumerSeekCallback { + + /** + * A callback method the user can implement to provide asynchronous handling of seek request completion. + * This method will be called when the fetch request sent to the server after a seek has been acknowledged. + * + * @param exception The exception thrown during processing of the request if the position to seek is out of range, + * or null if the seek completed successfully + */ + void onComplete(Exception exception); +} \ No newline at end of file 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 ed99e9bdf7c4e..70106487293b6 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 @@ -882,13 +882,35 @@ public void commit(CommitType commitType) { * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets + * + * @param partition The partition to apply the new fetch offset + * @param offset The fetch offset that the consumer will use for the partition */ @Override public void seek(TopicPartition partition, long offset) { + acquire(); + try { + seek(partition, offset, null); + } finally { + release(); + } + } + + /** + * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API + * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that + * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets + * + * @param partition The partition to apply the new fetch offset + * @param offset The fetch offset that the consumer will use for the partition + * @param callback Callback to be executed when the fetch request after the seek finishes + */ + @Override + public void seek(TopicPartition partition, long offset, ConsumerSeekCallback callback) { acquire(); try { log.debug("Seeking to offset {} for partition {}", offset, partition); - this.subscriptions.seek(partition, offset); + this.subscriptions.seek(partition, offset, callback); } finally { release(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index b07e760c28b17..cb10cfb37ebd5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -140,8 +140,13 @@ public synchronized void commit(CommitType commitType) { @Override public synchronized void seek(TopicPartition partition, long offset) { + seek(partition, offset, null); + } + + @Override + public synchronized void seek(TopicPartition partition, long offset, ConsumerSeekCallback callback) { ensureNotClosed(); - subscriptions.seek(partition, offset); + subscriptions.seek(partition, offset, callback); } @Override diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 9dc669728e6b0..7b158953f6b84 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -407,7 +407,7 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp); } else if (partition.errorCode == Errors.NONE.code()) { long fetchOffset = request.fetchData().get(tp).offset; - + subscriptions.invokeConsumerSeekCallback(tp, null); // we are interested in this fetch only if the beginning offset matches the // current consumed position Long consumed = subscriptions.consumed(tp); @@ -445,6 +445,7 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { // TODO: this could be optimized by grouping all out-of-range partitions log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); subscriptions.needOffsetReset(tp); + subscriptions.invokeConsumerSeekCallback(tp, Errors.OFFSET_OUT_OF_RANGE.exception()); } else if (partition.errorCode == Errors.UNKNOWN.code()) { log.warn("Unknown error fetching data for topic-partition {}", tp); } else { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index 6788ee67df8bb..ca4c1c604a360 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -12,6 +12,7 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.clients.consumer.ConsumerSeekCallback; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; @@ -159,7 +160,15 @@ public void commitsRefreshed() { } public void seek(TopicPartition tp, long offset) { - assignedState(tp).seek(offset); + seek(tp, offset, null); + } + + public void seek(TopicPartition tp, long offset, ConsumerSeekCallback callback) { + assignedState(tp).seek(offset, callback); + } + + public void invokeConsumerSeekCallback(TopicPartition tp, Exception exception) { + assignedState(tp).invokeConsumerSeekCallback(exception); } public Set assignedPartitions() { @@ -276,6 +285,8 @@ private static class TopicPartitionState { private boolean awaitingReset; // whether we are awaiting reset private OffsetResetStrategy resetStrategy; // the reset strategy if awaitingReset is set + private ConsumerSeekCallback consumerSeekCallback; // callback to be executed when the fetch request after the seek finishes + public TopicPartitionState() { this.paused = false; this.consumed = null; @@ -284,6 +295,7 @@ public TopicPartitionState() { this.awaitingReset = false; this.hasValidPosition = false; this.resetStrategy = null; + this.consumerSeekCallback = null; } private void awaitReset(OffsetResetStrategy strategy) { @@ -292,14 +304,16 @@ private void awaitReset(OffsetResetStrategy strategy) { this.consumed = null; this.fetched = null; this.hasValidPosition = false; + this.consumerSeekCallback = null; } - private void seek(long offset) { + private void seek(long offset, ConsumerSeekCallback callback) { this.consumed = offset; this.fetched = offset; this.awaitingReset = false; this.resetStrategy = null; this.hasValidPosition = true; + this.consumerSeekCallback = callback; } private void fetched(long offset) { @@ -314,6 +328,12 @@ private void consumed(long offset) { this.consumed = offset; } + private void invokeConsumerSeekCallback(Exception exception) { + if (consumerSeekCallback != null) + consumerSeekCallback.onComplete(exception); + consumerSeekCallback = null; + } + private void committed(Long offset) { this.committed = offset; } From d8912a23a638e4d40af2385c3322ffd911d49f19 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 5 Aug 2015 17:18:37 -0700 Subject: [PATCH 02/18] fix checkstyle issue --- .../consumer/ConsumerSeekCallback.java | 20 ++++++++----------- 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java index b4c619fbaba10..60061c35244dd 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java @@ -12,22 +12,18 @@ */ package org.apache.kafka.clients.consumer; -import org.apache.kafka.common.TopicPartition; - -import java.util.Map; - /** * A callback interface that the user can implement to trigger custom actions when a fetch request after a seek completes. The callback * may be executed in any thread calling {@link Consumer#poll(long) poll()}. */ public interface ConsumerSeekCallback { - /** - * A callback method the user can implement to provide asynchronous handling of seek request completion. - * This method will be called when the fetch request sent to the server after a seek has been acknowledged. - * - * @param exception The exception thrown during processing of the request if the position to seek is out of range, - * or null if the seek completed successfully - */ - void onComplete(Exception exception); + /** + * A callback method the user can implement to provide asynchronous handling of seek request completion. + * This method will be called when the fetch request sent to the server after a seek has been acknowledged. + * + * @param exception The exception thrown during processing of the request if the position to seek is out of range, + * or null if the seek completed successfully + */ + void onComplete(Exception exception); } \ No newline at end of file From 06ad6f8b1089c2e5a149921d512b06519e2e5974 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Tue, 11 Aug 2015 04:41:30 +0000 Subject: [PATCH 03/18] Always invoke ConsumerSeekCallback; only invoke the callback if fetched offset matches the seeked offset --- .../apache/kafka/clients/consumer/ConsumerSeekCallback.java | 3 ++- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 2 +- .../kafka/clients/consumer/internals/SubscriptionState.java | 5 ++++- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java index 60061c35244dd..d0802841415c8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java @@ -22,8 +22,9 @@ public interface ConsumerSeekCallback { * A callback method the user can implement to provide asynchronous handling of seek request completion. * This method will be called when the fetch request sent to the server after a seek has been acknowledged. * + * @param offset The offset used in the seek request * @param exception The exception thrown during processing of the request if the position to seek is out of range, * or null if the seek completed successfully */ - void onComplete(Exception exception); + void onComplete(Long offset, Exception exception); } \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 7b158953f6b84..3589628731be8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -407,7 +407,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp); } else if (partition.errorCode == Errors.NONE.code()) { long fetchOffset = request.fetchData().get(tp).offset; - subscriptions.invokeConsumerSeekCallback(tp, null); // we are interested in this fetch only if the beginning offset matches the // current consumed position Long consumed = subscriptions.consumed(tp); @@ -421,6 +420,7 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { continue; } + subscriptions.invokeConsumerSeekCallback(tp, null); int bytes = 0; ByteBuffer buffer = partition.recordSet; MemoryRecords records = MemoryRecords.readableRecords(buffer); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index ca4c1c604a360..c4adf763dcb58 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -112,10 +112,13 @@ public void unsubscribe(TopicPartition partition) { } private void clearPartition(TopicPartition tp) { + invokeConsumerSeekCallback(tp, null); this.assignedPartitions.remove(tp); } public void clearAssignment() { + for (TopicPartition tp: this.assignedPartitions.keySet()) + invokeConsumerSeekCallback(tp, null); this.assignedPartitions.clear(); this.needsPartitionAssignment = !subscribedTopics().isEmpty(); } @@ -330,7 +333,7 @@ private void consumed(long offset) { private void invokeConsumerSeekCallback(Exception exception) { if (consumerSeekCallback != null) - consumerSeekCallback.onComplete(exception); + consumerSeekCallback.onComplete(this.fetched, exception); consumerSeekCallback = null; } From 4c3cca2e8da5d657373bfd591a45cd190d394d96 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Sun, 16 Aug 2015 23:24:19 +0000 Subject: [PATCH 04/18] Change callback name and API document of ConsumerSeekCallback to better explain its usage --- .../apache/kafka/clients/consumer/ConsumerSeekCallback.java | 6 +++++- .../kafka/clients/consumer/internals/SubscriptionState.java | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java index d0802841415c8..67c560c6d8b93 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java @@ -22,9 +22,13 @@ public interface ConsumerSeekCallback { * A callback method the user can implement to provide asynchronous handling of seek request completion. * This method will be called when the fetch request sent to the server after a seek has been acknowledged. * + * Upon KafkaConsumer#seek(TopicPartition, long, ConsumerSeekCallback) the offset of the topicParition will be updated + * immediately for use in the next fetch request. But we can only find out whether the offset is valid when we receive + * response from server corresponding to the fetch request after the seek. Therefore this callback will be called asynchronously. + * * @param offset The offset used in the seek request * @param exception The exception thrown during processing of the request if the position to seek is out of range, * or null if the seek completed successfully */ - void onComplete(Long offset, Exception exception); + void onResponse(Long offset, Exception exception); } \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index c4adf763dcb58..5ac041c4e4e5e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -333,7 +333,7 @@ private void consumed(long offset) { private void invokeConsumerSeekCallback(Exception exception) { if (consumerSeekCallback != null) - consumerSeekCallback.onComplete(this.fetched, exception); + consumerSeekCallback.onResponse(this.fetched, exception); consumerSeekCallback = null; } From 08ee9bd32a535bad9367bfede75e3b226e3cb4dc Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Mon, 17 Aug 2015 17:18:48 +0000 Subject: [PATCH 05/18] use a more intuitive name for ConsumerSeekCallback function --- .../org/apache/kafka/clients/consumer/ConsumerSeekCallback.java | 2 +- .../kafka/clients/consumer/internals/SubscriptionState.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java index 67c560c6d8b93..5db46d5d55029 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java @@ -30,5 +30,5 @@ public interface ConsumerSeekCallback { * @param exception The exception thrown during processing of the request if the position to seek is out of range, * or null if the seek completed successfully */ - void onResponse(Long offset, Exception exception); + void onFirstFetchResponse(Long offset, Exception exception); } \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index 5ac041c4e4e5e..bbd757faafeac 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -333,7 +333,7 @@ private void consumed(long offset) { private void invokeConsumerSeekCallback(Exception exception) { if (consumerSeekCallback != null) - consumerSeekCallback.onResponse(this.fetched, exception); + consumerSeekCallback.onFirstFetchResponse(this.fetched, exception); consumerSeekCallback = null; } From 1d8cb21d918f0eb16e53d4dec0d9dac4f2990263 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 2 Sep 2015 18:40:51 +0000 Subject: [PATCH 06/18] undo all changes made earlier --- .../kafka/clients/consumer/Consumer.java | 5 --- .../consumer/ConsumerSeekCallback.java | 34 ------------------- .../kafka/clients/consumer/KafkaConsumer.java | 24 +------------ .../kafka/clients/consumer/MockConsumer.java | 7 +--- .../clients/consumer/internals/Fetcher.java | 3 +- .../consumer/internals/SubscriptionState.java | 27 ++------------- 6 files changed, 5 insertions(+), 95 deletions(-) delete mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java index 1e052915afcea..158e1ea1e35db 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java @@ -83,11 +83,6 @@ public interface Consumer extends Closeable { */ public void seek(TopicPartition partition, long offset); - /** - * @see KafkaConsumer#seek(TopicPartition, long, ConsumerSeekCallback) - */ - public void seek(TopicPartition partition, long offset, ConsumerSeekCallback callbak); - /** * @see KafkaConsumer#seekToBeginning(TopicPartition...) */ diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java deleted file mode 100644 index 5db46d5d55029..0000000000000 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerSeekCallback.java +++ /dev/null @@ -1,34 +0,0 @@ -/** - * 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.kafka.clients.consumer; - -/** - * A callback interface that the user can implement to trigger custom actions when a fetch request after a seek completes. The callback - * may be executed in any thread calling {@link Consumer#poll(long) poll()}. - */ -public interface ConsumerSeekCallback { - - /** - * A callback method the user can implement to provide asynchronous handling of seek request completion. - * This method will be called when the fetch request sent to the server after a seek has been acknowledged. - * - * Upon KafkaConsumer#seek(TopicPartition, long, ConsumerSeekCallback) the offset of the topicParition will be updated - * immediately for use in the next fetch request. But we can only find out whether the offset is valid when we receive - * response from server corresponding to the fetch request after the seek. Therefore this callback will be called asynchronously. - * - * @param offset The offset used in the seek request - * @param exception The exception thrown during processing of the request if the position to seek is out of range, - * or null if the seek completed successfully - */ - void onFirstFetchResponse(Long offset, Exception exception); -} \ No newline at end of file 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 70106487293b6..ed99e9bdf7c4e 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 @@ -882,35 +882,13 @@ public void commit(CommitType commitType) { * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets - * - * @param partition The partition to apply the new fetch offset - * @param offset The fetch offset that the consumer will use for the partition */ @Override public void seek(TopicPartition partition, long offset) { - acquire(); - try { - seek(partition, offset, null); - } finally { - release(); - } - } - - /** - * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API - * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that - * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets - * - * @param partition The partition to apply the new fetch offset - * @param offset The fetch offset that the consumer will use for the partition - * @param callback Callback to be executed when the fetch request after the seek finishes - */ - @Override - public void seek(TopicPartition partition, long offset, ConsumerSeekCallback callback) { acquire(); try { log.debug("Seeking to offset {} for partition {}", offset, partition); - this.subscriptions.seek(partition, offset, callback); + this.subscriptions.seek(partition, offset); } finally { release(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index cb10cfb37ebd5..b07e760c28b17 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -140,13 +140,8 @@ public synchronized void commit(CommitType commitType) { @Override public synchronized void seek(TopicPartition partition, long offset) { - seek(partition, offset, null); - } - - @Override - public synchronized void seek(TopicPartition partition, long offset, ConsumerSeekCallback callback) { ensureNotClosed(); - subscriptions.seek(partition, offset, callback); + subscriptions.seek(partition, offset); } @Override diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 3589628731be8..9dc669728e6b0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -407,6 +407,7 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp); } else if (partition.errorCode == Errors.NONE.code()) { long fetchOffset = request.fetchData().get(tp).offset; + // we are interested in this fetch only if the beginning offset matches the // current consumed position Long consumed = subscriptions.consumed(tp); @@ -420,7 +421,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { continue; } - subscriptions.invokeConsumerSeekCallback(tp, null); int bytes = 0; ByteBuffer buffer = partition.recordSet; MemoryRecords records = MemoryRecords.readableRecords(buffer); @@ -445,7 +445,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { // TODO: this could be optimized by grouping all out-of-range partitions log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); subscriptions.needOffsetReset(tp); - subscriptions.invokeConsumerSeekCallback(tp, Errors.OFFSET_OUT_OF_RANGE.exception()); } else if (partition.errorCode == Errors.UNKNOWN.code()) { log.warn("Unknown error fetching data for topic-partition {}", tp); } else { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index bbd757faafeac..6788ee67df8bb 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -12,7 +12,6 @@ */ package org.apache.kafka.clients.consumer.internals; -import org.apache.kafka.clients.consumer.ConsumerSeekCallback; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; @@ -112,13 +111,10 @@ public void unsubscribe(TopicPartition partition) { } private void clearPartition(TopicPartition tp) { - invokeConsumerSeekCallback(tp, null); this.assignedPartitions.remove(tp); } public void clearAssignment() { - for (TopicPartition tp: this.assignedPartitions.keySet()) - invokeConsumerSeekCallback(tp, null); this.assignedPartitions.clear(); this.needsPartitionAssignment = !subscribedTopics().isEmpty(); } @@ -163,15 +159,7 @@ public void commitsRefreshed() { } public void seek(TopicPartition tp, long offset) { - seek(tp, offset, null); - } - - public void seek(TopicPartition tp, long offset, ConsumerSeekCallback callback) { - assignedState(tp).seek(offset, callback); - } - - public void invokeConsumerSeekCallback(TopicPartition tp, Exception exception) { - assignedState(tp).invokeConsumerSeekCallback(exception); + assignedState(tp).seek(offset); } public Set assignedPartitions() { @@ -288,8 +276,6 @@ private static class TopicPartitionState { private boolean awaitingReset; // whether we are awaiting reset private OffsetResetStrategy resetStrategy; // the reset strategy if awaitingReset is set - private ConsumerSeekCallback consumerSeekCallback; // callback to be executed when the fetch request after the seek finishes - public TopicPartitionState() { this.paused = false; this.consumed = null; @@ -298,7 +284,6 @@ public TopicPartitionState() { this.awaitingReset = false; this.hasValidPosition = false; this.resetStrategy = null; - this.consumerSeekCallback = null; } private void awaitReset(OffsetResetStrategy strategy) { @@ -307,16 +292,14 @@ private void awaitReset(OffsetResetStrategy strategy) { this.consumed = null; this.fetched = null; this.hasValidPosition = false; - this.consumerSeekCallback = null; } - private void seek(long offset, ConsumerSeekCallback callback) { + private void seek(long offset) { this.consumed = offset; this.fetched = offset; this.awaitingReset = false; this.resetStrategy = null; this.hasValidPosition = true; - this.consumerSeekCallback = callback; } private void fetched(long offset) { @@ -331,12 +314,6 @@ private void consumed(long offset) { this.consumed = offset; } - private void invokeConsumerSeekCallback(Exception exception) { - if (consumerSeekCallback != null) - consumerSeekCallback.onFirstFetchResponse(this.fetched, exception); - consumerSeekCallback = null; - } - private void committed(Long offset) { this.committed = offset; } From 365033466bd21cbe7af25b8dcb074dd1c2e577df Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 2 Sep 2015 22:50:10 +0000 Subject: [PATCH 07/18] KAFKA-2390; OffsetOutOfRangeException should contain the Offset and Partition info. --- .../kafka/clients/consumer/KafkaConsumer.java | 2 +- .../clients/consumer/internals/Fetcher.java | 35 ++++++++++++++++++- .../errors/OffsetOutOfRangeException.java | 12 +++++++ .../consumer/internals/FetcherTest.java | 16 ++++----- 4 files changed, 55 insertions(+), 10 deletions(-) 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 73237e455a9e5..3b9c4bcf93abd 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 @@ -769,7 +769,7 @@ private Map>> pollOnce(long timeout) { Cluster cluster = this.metadata.fetch(); fetcher.initFetches(cluster); client.poll(timeout); - return fetcher.fetchedRecords(); + return fetcher.fetchedRecordsOrException(); } private void scheduleAutoCommitTask(final long interval) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 1ae6d03f23628..b3bdf20a14d6e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -25,6 +25,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.DisconnectException; import org.apache.kafka.common.errors.InvalidMetadataException; +import org.apache.kafka.common.errors.OffsetOutOfRangeException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; @@ -80,6 +81,8 @@ public class Fetcher { private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; + private OffsetOutOfRangeException exceptionFromLastFetch; + public Fetcher(ConsumerNetworkClient client, int minBytes, int maxWaitMs, @@ -108,6 +111,7 @@ public Fetcher(ConsumerNetworkClient client, this.valueDeserializer = valueDeserializer; this.records = new LinkedList>(); + this.exceptionFromLastFetch = null; this.sensors = new FetchManagerMetrics(metrics, metricGrpPrefix, metricTags); this.retryBackoffMs = retryBackoffMs; @@ -139,6 +143,7 @@ public void onFailure(RuntimeException e) { /** * Update the fetch positions for the provided partitions. * @param partitions + * @throws NoOffsetForPartitionException If no offset is stored for a given partition and no reset policy is available */ public void updateFetchPositions(Set partitions) { // reset the fetch position to the committed position @@ -256,12 +261,19 @@ private long listOffset(TopicPartition partition, long timestamp) { * Return the fetched records, empty the record buffer and update the consumed position. * * @return The fetched records per partition + * @throws OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse */ - public Map>> fetchedRecords() { + public Map>> fetchedRecordsOrException() { if (this.subscriptions.partitionAssignmentNeeded()) { return Collections.emptyMap(); } else { Map>> drained = new HashMap<>(); + if (exceptionFromLastFetch != null) { + RuntimeException e = this.exceptionFromLastFetch; + this.exceptionFromLastFetch = null; + throw e; + } + for (PartitionRecords part : this.records) { if (!subscriptions.isFetchable(part.partition)) { log.debug("Ignoring fetched records for {} since it is no longer fetchable", part.partition); @@ -400,12 +412,33 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { int totalBytes = 0; int totalCount = 0; FetchResponse response = new FetchResponse(resp.responseBody()); + + // Look for OffsetOutOfRange error in fetchResponse + Map offsetOutOfRangePartitions = new HashMap(); + for (Map.Entry entry : response.responseData().entrySet()) { + TopicPartition tp = entry.getKey(); + FetchResponse.PartitionData partition = entry.getValue(); + long fetchOffset = request.fetchData().get(tp).offset; + if (subscriptions.assignedPartitions().contains(tp) && partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) + offsetOutOfRangePartitions.put(tp, fetchOffset); + } + if (!offsetOutOfRangePartitions.isEmpty()) + this.exceptionFromLastFetch = new OffsetOutOfRangeException(offsetOutOfRangePartitions); + for (Map.Entry entry : response.responseData().entrySet()) { TopicPartition tp = entry.getKey(); FetchResponse.PartitionData partition = entry.getValue(); if (!subscriptions.assignedPartitions().contains(tp)) { log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp); } else if (partition.errorCode == Errors.NONE.code()) { + + /** + * Don't read data if there is OffsetOutOfRangeException for any partition in the fetchResponse. + * We will fetch the data for all partitions in the fetchRequest again. + */ + if (this.exceptionFromLastFetch != null) + continue; + long fetchOffset = request.fetchData().get(tp).offset; // we are interested in this fetch only if the beginning offset matches the diff --git a/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java b/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java index fc7c6e3471b05..16f2e186d465a 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java @@ -12,6 +12,9 @@ */ package org.apache.kafka.common.errors; +import org.apache.kafka.common.TopicPartition; +import java.util.Map; + /** * This offset is either larger or smaller than the range of offsets the server has for the given partition. * @@ -19,10 +22,15 @@ public class OffsetOutOfRangeException extends RetriableException { private static final long serialVersionUID = 1L; + Map offsetOutOfRangePartitions = null; public OffsetOutOfRangeException() { } + public OffsetOutOfRangeException(Map offsetOutOfRangePartitions) { + this.offsetOutOfRangePartitions = offsetOutOfRangePartitions; + } + public OffsetOutOfRangeException(String message) { super(message); } @@ -35,4 +43,8 @@ public OffsetOutOfRangeException(String message, Throwable cause) { super(message, cause); } + public Map getOutOfRangePartitions() { + return offsetOutOfRangePartitions; + } + } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index f2a8381c11ca8..8c8468e6afdc9 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -115,7 +115,7 @@ public void testFetchNormal() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0)); consumerClient.poll(0); - records = fetcher.fetchedRecords().get(tp); + records = fetcher.fetchedRecordsOrException().get(tp); assertEquals(3, records.size()); assertEquals(4L, (long) subscriptions.fetched(tp)); // this is the next fetching position assertEquals(4L, (long) subscriptions.consumed(tp)); @@ -140,7 +140,7 @@ public void testFetchDuringRebalance() { consumerClient.poll(0); // The active fetch should be ignored since its position is no longer valid - assertTrue(fetcher.fetchedRecords().isEmpty()); + assertTrue(fetcher.fetchedRecordsOrException().isEmpty()); } @Test @@ -153,7 +153,7 @@ public void testInFlightFetchOnPausedPartition() { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0)); consumerClient.poll(0); - assertNull(fetcher.fetchedRecords().get(tp)); + assertNull(fetcher.fetchedRecordsOrException().get(tp)); } @Test @@ -174,7 +174,7 @@ public void testFetchNotLeaderForPartition() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NOT_LEADER_FOR_PARTITION.code(), 100L, 0)); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecordsOrException().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -186,7 +186,7 @@ public void testFetchUnknownTopicOrPartition() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.UNKNOWN_TOPIC_OR_PARTITION.code(), 100L, 0)); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecordsOrException().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -199,7 +199,7 @@ public void testFetchOffsetOutOfRange() { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.OFFSET_OUT_OF_RANGE.code(), 100L, 0)); consumerClient.poll(0); assertTrue(subscriptions.isOffsetResetNeeded(tp)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecordsOrException().size()); assertEquals(null, subscriptions.fetched(tp)); assertEquals(null, subscriptions.consumed(tp)); } @@ -212,7 +212,7 @@ public void testFetchDisconnected() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0), true); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecordsOrException().size()); // disconnects should have no affect on subscription state assertFalse(subscriptions.isOffsetResetNeeded(tp)); @@ -321,7 +321,7 @@ public void testQuotaMetrics() throws Exception { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 100 * i)); consumerClient.poll(0); - records = fetcher.fetchedRecords().get(tp); + records = fetcher.fetchedRecordsOrException().get(tp); assertEquals(3, records.size()); } From 83872ae2e8201ce4b4be5b9cea0cb56718946615 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Thu, 3 Sep 2015 00:35:59 +0000 Subject: [PATCH 08/18] throw exception only if defaultResetStrategy != None --- .../kafka/clients/consumer/KafkaConsumer.java | 2 +- .../clients/consumer/internals/Fetcher.java | 15 ++++++++------- .../consumer/internals/SubscriptionState.java | 4 ++++ .../clients/consumer/internals/FetcherTest.java | 16 ++++++++-------- 4 files changed, 21 insertions(+), 16 deletions(-) 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 3b9c4bcf93abd..73237e455a9e5 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 @@ -769,7 +769,7 @@ private Map>> pollOnce(long timeout) { Cluster cluster = this.metadata.fetch(); fetcher.initFetches(cluster); client.poll(timeout); - return fetcher.fetchedRecordsOrException(); + return fetcher.fetchedRecords(); } private void scheduleAutoCommitTask(final long interval) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index b3bdf20a14d6e..2ee683a4fe315 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -263,7 +263,7 @@ private long listOffset(TopicPartition partition, long timestamp) { * @return The fetched records per partition * @throws OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse */ - public Map>> fetchedRecordsOrException() { + public Map>> fetchedRecords() { if (this.subscriptions.partitionAssignmentNeeded()) { return Collections.emptyMap(); } else { @@ -419,8 +419,13 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { TopicPartition tp = entry.getKey(); FetchResponse.PartitionData partition = entry.getValue(); long fetchOffset = request.fetchData().get(tp).offset; - if (subscriptions.assignedPartitions().contains(tp) && partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) - offsetOutOfRangePartitions.put(tp, fetchOffset); + if (subscriptions.assignedPartitions().contains(tp) && partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { + log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); + if (subscriptions.hasDefaultOffsetResetPolicy()) + subscriptions.needOffsetReset(tp); + else + offsetOutOfRangePartitions.put(tp, fetchOffset); + } } if (!offsetOutOfRangePartitions.isEmpty()) this.exceptionFromLastFetch = new OffsetOutOfRangeException(offsetOutOfRangePartitions); @@ -474,10 +479,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { } else if (partition.errorCode == Errors.NOT_LEADER_FOR_PARTITION.code() || partition.errorCode == Errors.UNKNOWN_TOPIC_OR_PARTITION.code()) { this.metadata.requestUpdate(); - } else if (partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { - // TODO: this could be optimized by grouping all out-of-range partitions - log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); - subscriptions.needOffsetReset(tp); } else if (partition.errorCode == Errors.UNKNOWN.code()) { log.warn("Unknown error fetching data for topic-partition {}", tp); } else { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index ec6b424bc256f..16c9b7afa65ae 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -209,6 +209,10 @@ public void needOffsetReset(TopicPartition partition) { needOffsetReset(partition, defaultResetStrategy); } + public boolean hasDefaultOffsetResetPolicy() { + return defaultResetStrategy != OffsetResetStrategy.NONE; + } + public boolean isOffsetResetNeeded(TopicPartition partition) { return assignedState(partition).awaitingReset; } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index 8c8468e6afdc9..f2a8381c11ca8 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -115,7 +115,7 @@ public void testFetchNormal() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0)); consumerClient.poll(0); - records = fetcher.fetchedRecordsOrException().get(tp); + records = fetcher.fetchedRecords().get(tp); assertEquals(3, records.size()); assertEquals(4L, (long) subscriptions.fetched(tp)); // this is the next fetching position assertEquals(4L, (long) subscriptions.consumed(tp)); @@ -140,7 +140,7 @@ public void testFetchDuringRebalance() { consumerClient.poll(0); // The active fetch should be ignored since its position is no longer valid - assertTrue(fetcher.fetchedRecordsOrException().isEmpty()); + assertTrue(fetcher.fetchedRecords().isEmpty()); } @Test @@ -153,7 +153,7 @@ public void testInFlightFetchOnPausedPartition() { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0)); consumerClient.poll(0); - assertNull(fetcher.fetchedRecordsOrException().get(tp)); + assertNull(fetcher.fetchedRecords().get(tp)); } @Test @@ -174,7 +174,7 @@ public void testFetchNotLeaderForPartition() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NOT_LEADER_FOR_PARTITION.code(), 100L, 0)); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecordsOrException().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -186,7 +186,7 @@ public void testFetchUnknownTopicOrPartition() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.UNKNOWN_TOPIC_OR_PARTITION.code(), 100L, 0)); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecordsOrException().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -199,7 +199,7 @@ public void testFetchOffsetOutOfRange() { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.OFFSET_OUT_OF_RANGE.code(), 100L, 0)); consumerClient.poll(0); assertTrue(subscriptions.isOffsetResetNeeded(tp)); - assertEquals(0, fetcher.fetchedRecordsOrException().size()); + assertEquals(0, fetcher.fetchedRecords().size()); assertEquals(null, subscriptions.fetched(tp)); assertEquals(null, subscriptions.consumed(tp)); } @@ -212,7 +212,7 @@ public void testFetchDisconnected() { fetcher.initFetches(cluster); client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 0), true); consumerClient.poll(0); - assertEquals(0, fetcher.fetchedRecordsOrException().size()); + assertEquals(0, fetcher.fetchedRecords().size()); // disconnects should have no affect on subscription state assertFalse(subscriptions.isOffsetResetNeeded(tp)); @@ -321,7 +321,7 @@ public void testQuotaMetrics() throws Exception { client.prepareResponse(fetchResponse(this.records.buffer(), Errors.NONE.code(), 100L, 100 * i)); consumerClient.poll(0); - records = fetcher.fetchedRecordsOrException().get(tp); + records = fetcher.fetchedRecords().get(tp); assertEquals(3, records.size()); } From a3eb97b2bf12ba75feeccb752d679148b479fdbf Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Thu, 3 Sep 2015 02:08:24 +0000 Subject: [PATCH 09/18] only throw exception for partitions that are still fetchable --- .../clients/consumer/internals/Fetcher.java | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 2ee683a4fe315..4f4daefff7de8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -81,7 +81,7 @@ public class Fetcher { private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; - private OffsetOutOfRangeException exceptionFromLastFetch; + private Map offsetOutOfRangePartitions; public Fetcher(ConsumerNetworkClient client, int minBytes, @@ -111,7 +111,7 @@ public Fetcher(ConsumerNetworkClient client, this.valueDeserializer = valueDeserializer; this.records = new LinkedList>(); - this.exceptionFromLastFetch = null; + this.offsetOutOfRangePartitions = new HashMap<>(); this.sensors = new FetchManagerMetrics(metrics, metricGrpPrefix, metricTags); this.retryBackoffMs = retryBackoffMs; @@ -268,11 +268,20 @@ public Map>> fetchedRecords() { return Collections.emptyMap(); } else { Map>> drained = new HashMap<>(); - if (exceptionFromLastFetch != null) { - RuntimeException e = this.exceptionFromLastFetch; - this.exceptionFromLastFetch = null; - throw e; + Map currentOutOfRangePartitions = new HashMap<>(); + + // filter offsetOutOfRangePartitions to retain only the fetchable partitions + for (Map.Entry entry: this.offsetOutOfRangePartitions.entrySet()) { + if (!subscriptions.isFetchable(entry.getKey())) { + log.debug("Ignoring fetched records for {} since it is no longer fetchable", entry.getKey()); + continue; + } + currentOutOfRangePartitions.put(entry.getKey(), entry.getValue()); } + this.offsetOutOfRangePartitions.clear(); + if (!currentOutOfRangePartitions.isEmpty()) + throw new OffsetOutOfRangeException(currentOutOfRangePartitions); + for (PartitionRecords part : this.records) { if (!subscriptions.isFetchable(part.partition)) { @@ -414,7 +423,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { FetchResponse response = new FetchResponse(resp.responseBody()); // Look for OffsetOutOfRange error in fetchResponse - Map offsetOutOfRangePartitions = new HashMap(); for (Map.Entry entry : response.responseData().entrySet()) { TopicPartition tp = entry.getKey(); FetchResponse.PartitionData partition = entry.getValue(); @@ -424,11 +432,9 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { if (subscriptions.hasDefaultOffsetResetPolicy()) subscriptions.needOffsetReset(tp); else - offsetOutOfRangePartitions.put(tp, fetchOffset); + this.offsetOutOfRangePartitions.put(tp, fetchOffset); } } - if (!offsetOutOfRangePartitions.isEmpty()) - this.exceptionFromLastFetch = new OffsetOutOfRangeException(offsetOutOfRangePartitions); for (Map.Entry entry : response.responseData().entrySet()) { TopicPartition tp = entry.getKey(); @@ -441,7 +447,7 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { * Don't read data if there is OffsetOutOfRangeException for any partition in the fetchResponse. * We will fetch the data for all partitions in the fetchRequest again. */ - if (this.exceptionFromLastFetch != null) + if (!this.offsetOutOfRangePartitions.isEmpty()) continue; long fetchOffset = request.fetchData().get(tp).offset; From bc75c78c73cf8c16b3c31e6196440eaf4ba1894e Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Thu, 3 Sep 2015 12:16:44 -0700 Subject: [PATCH 10/18] Don't initiate fetch if we already have records available from previous fetchRequest --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 4f4daefff7de8..ee836fc0026a3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -123,6 +123,11 @@ public Fetcher(ConsumerNetworkClient client, * @param cluster The current cluster metadata */ public void initFetches(Cluster cluster) { + // Don't initiate fetch if we already have records available from previous fetchRequest + // This may happen when there is OffsetOutOfRange exception from previous fetchRequest + if (!records.isEmpty()) + return; + for (Map.Entry fetchEntry: createFetchRequests(cluster).entrySet()) { final FetchRequest fetch = fetchEntry.getValue(); client.send(fetchEntry.getKey(), ApiKeys.FETCH, fetch) From 12e84b5cdff994806698a988bdf3118228d90746 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Fri, 4 Sep 2015 17:55:20 +0000 Subject: [PATCH 11/18] fix test failure --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index ee836fc0026a3..3add120d4fbda 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -433,7 +433,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { FetchResponse.PartitionData partition = entry.getValue(); long fetchOffset = request.fetchData().get(tp).offset; if (subscriptions.assignedPartitions().contains(tp) && partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { - log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); if (subscriptions.hasDefaultOffsetResetPolicy()) subscriptions.needOffsetReset(tp); else @@ -492,6 +491,8 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { this.metadata.requestUpdate(); } else if (partition.errorCode == Errors.UNKNOWN.code()) { log.warn("Unknown error fetching data for topic-partition {}", tp); + } else if (partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { + log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); } else { throw new IllegalStateException("Unexpected error code " + partition.errorCode + " while fetching data"); } From fb75a11ed70f674e7594555440fdb889d55c349c Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Fri, 4 Sep 2015 22:25:58 +0000 Subject: [PATCH 12/18] fix code to address reviews --- .../clients/consumer/internals/Fetcher.java | 65 +++++++++---------- 1 file changed, 30 insertions(+), 35 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 3add120d4fbda..fdd2e231023e0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -262,6 +262,30 @@ private long listOffset(TopicPartition partition, long timestamp) { } } + /** + * If any partition from previous fetchResponse contains OffsetOutOfRange error, throw + * OffsetOutOfRangeException and clear the partition list. + * + * @throws OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse + */ + private void throwIfOffsetOutOfRange() throws OffsetOutOfRangeException{ + Map currentOutOfRangePartitions = new HashMap<>(); + + // filter offsetOutOfRangePartitions to retain only the fetchable partitions + for (Map.Entry entry: this.offsetOutOfRangePartitions.entrySet()) { + if (!subscriptions.isFetchable(entry.getKey())) { + log.debug("Ignoring fetched records for {} since it is no longer fetchable", entry.getKey()); + continue; + } + Long consumed = subscriptions.consumed(entry.getKey()); + if (consumed != null && entry.getValue() == consumed) + currentOutOfRangePartitions.put(entry.getKey(), entry.getValue()); + } + this.offsetOutOfRangePartitions.clear(); + if (!currentOutOfRangePartitions.isEmpty()) + throw new OffsetOutOfRangeException(currentOutOfRangePartitions); + } + /** * Return the fetched records, empty the record buffer and update the consumed position. * @@ -273,20 +297,7 @@ public Map>> fetchedRecords() { return Collections.emptyMap(); } else { Map>> drained = new HashMap<>(); - Map currentOutOfRangePartitions = new HashMap<>(); - - // filter offsetOutOfRangePartitions to retain only the fetchable partitions - for (Map.Entry entry: this.offsetOutOfRangePartitions.entrySet()) { - if (!subscriptions.isFetchable(entry.getKey())) { - log.debug("Ignoring fetched records for {} since it is no longer fetchable", entry.getKey()); - continue; - } - currentOutOfRangePartitions.put(entry.getKey(), entry.getValue()); - } - this.offsetOutOfRangePartitions.clear(); - if (!currentOutOfRangePartitions.isEmpty()) - throw new OffsetOutOfRangeException(currentOutOfRangePartitions); - + throwIfOffsetOutOfRange(); for (PartitionRecords part : this.records) { if (!subscriptions.isFetchable(part.partition)) { @@ -427,33 +438,12 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { int totalCount = 0; FetchResponse response = new FetchResponse(resp.responseBody()); - // Look for OffsetOutOfRange error in fetchResponse - for (Map.Entry entry : response.responseData().entrySet()) { - TopicPartition tp = entry.getKey(); - FetchResponse.PartitionData partition = entry.getValue(); - long fetchOffset = request.fetchData().get(tp).offset; - if (subscriptions.assignedPartitions().contains(tp) && partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { - if (subscriptions.hasDefaultOffsetResetPolicy()) - subscriptions.needOffsetReset(tp); - else - this.offsetOutOfRangePartitions.put(tp, fetchOffset); - } - } - for (Map.Entry entry : response.responseData().entrySet()) { TopicPartition tp = entry.getKey(); FetchResponse.PartitionData partition = entry.getValue(); if (!subscriptions.assignedPartitions().contains(tp)) { log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp); } else if (partition.errorCode == Errors.NONE.code()) { - - /** - * Don't read data if there is OffsetOutOfRangeException for any partition in the fetchResponse. - * We will fetch the data for all partitions in the fetchRequest again. - */ - if (!this.offsetOutOfRangePartitions.isEmpty()) - continue; - long fetchOffset = request.fetchData().get(tp).offset; // we are interested in this fetch only if the beginning offset matches the @@ -492,6 +482,11 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { } else if (partition.errorCode == Errors.UNKNOWN.code()) { log.warn("Unknown error fetching data for topic-partition {}", tp); } else if (partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { + long fetchOffset = request.fetchData().get(tp).offset; + if (subscriptions.hasDefaultOffsetResetPolicy()) + subscriptions.needOffsetReset(tp); + else + this.offsetOutOfRangePartitions.put(tp, fetchOffset); log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); } else { throw new IllegalStateException("Unexpected error code " + partition.errorCode + " while fetching data"); From 4fb44f97f8a2657d41d4678ce3aa63314ef351ff Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Fri, 4 Sep 2015 22:35:16 +0000 Subject: [PATCH 13/18] minor fix --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index fdd2e231023e0..ea4cc703d7e27 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -268,7 +268,7 @@ private long listOffset(TopicPartition partition, long timestamp) { * * @throws OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse */ - private void throwIfOffsetOutOfRange() throws OffsetOutOfRangeException{ + private void throwIfOffsetOutOfRange() throws OffsetOutOfRangeException { Map currentOutOfRangePartitions = new HashMap<>(); // filter offsetOutOfRangePartitions to retain only the fetchable partitions From f8a92b3cd8b4eb9e520f7e2e701b8f4cb868eb3b Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Thu, 10 Sep 2015 22:57:07 +0000 Subject: [PATCH 14/18] only create fetchRequest for partitions whose consumed == fetched offset --- .../org/apache/kafka/clients/consumer/KafkaConsumer.java | 5 +++++ .../apache/kafka/clients/consumer/internals/Fetcher.java | 9 +++------ 2 files changed, 8 insertions(+), 6 deletions(-) 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 73237e455a9e5..bd1f6c7942e3c 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 @@ -767,6 +767,11 @@ private Map>> pollOnce(long timeout) { // init any new fetches (won't resend pending fetches) Cluster cluster = this.metadata.fetch(); + Map records = fetcher.fetchedRecords(); + if (!records.isEmpty()) { + client.poll(0); + return records; + } fetcher.initFetches(cluster); client.poll(timeout); return fetcher.fetchedRecords(); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index ea4cc703d7e27..e062cd62baf1b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -123,11 +123,6 @@ public Fetcher(ConsumerNetworkClient client, * @param cluster The current cluster metadata */ public void initFetches(Cluster cluster) { - // Don't initiate fetch if we already have records available from previous fetchRequest - // This may happen when there is OffsetOutOfRange exception from previous fetchRequest - if (!records.isEmpty()) - return; - for (Map.Entry fetchEntry: createFetchRequests(cluster).entrySet()) { final FetchRequest fetch = fetchEntry.getValue(); client.send(fetchEntry.getKey(), ApiKeys.FETCH, fetch) @@ -411,7 +406,9 @@ private Map createFetchRequests(Cluster cluster) { } long offset = this.subscriptions.fetched(partition); - fetch.put(partition, new FetchRequest.PartitionData(offset, this.fetchSize)); + long consumed = this.subscriptions.consumed(partition); + if (consumed == offset) + fetch.put(partition, new FetchRequest.PartitionData(offset, this.fetchSize)); } } From 94b2ca096d1f4aa2d423bdefa053ec77f0370417 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Thu, 10 Sep 2015 23:03:35 +0000 Subject: [PATCH 15/18] remove unnecessary empty line and line movement --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index e062cd62baf1b..b90f1df9d7dae 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -434,7 +434,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { int totalBytes = 0; int totalCount = 0; FetchResponse response = new FetchResponse(resp.responseBody()); - for (Map.Entry entry : response.responseData().entrySet()) { TopicPartition tp = entry.getKey(); FetchResponse.PartitionData partition = entry.getValue(); @@ -476,8 +475,6 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { } else if (partition.errorCode == Errors.NOT_LEADER_FOR_PARTITION.code() || partition.errorCode == Errors.UNKNOWN_TOPIC_OR_PARTITION.code()) { this.metadata.requestUpdate(); - } else if (partition.errorCode == Errors.UNKNOWN.code()) { - log.warn("Unknown error fetching data for topic-partition {}", tp); } else if (partition.errorCode == Errors.OFFSET_OUT_OF_RANGE.code()) { long fetchOffset = request.fetchData().get(tp).offset; if (subscriptions.hasDefaultOffsetResetPolicy()) @@ -485,6 +482,8 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { else this.offsetOutOfRangePartitions.put(tp, fetchOffset); log.info("Fetch offset {} is out of range, resetting offset", subscriptions.fetched(tp)); + } else if (partition.errorCode == Errors.UNKNOWN.code()) { + log.warn("Unknown error fetching data for topic-partition {}", tp); } else { throw new IllegalStateException("Unexpected error code " + partition.errorCode + " while fetching data"); } From 176ef240352a0ca2586a0752ebb67cf385f5ef3f Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Fri, 11 Sep 2015 21:54:33 +0000 Subject: [PATCH 16/18] add comments --- .../org/apache/kafka/clients/consumer/KafkaConsumer.java | 3 ++- .../apache/kafka/clients/consumer/internals/Fetcher.java | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) 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 bd1f6c7942e3c..52a80c4b6dac6 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 @@ -767,7 +767,8 @@ private Map>> pollOnce(long timeout) { // init any new fetches (won't resend pending fetches) Cluster cluster = this.metadata.fetch(); - Map records = fetcher.fetchedRecords(); + Map>> records = fetcher.fetchedRecords(); + // Avoid block waiting for response if we already have data available, e.g. from another API call to commit. if (!records.isEmpty()) { client.poll(0); return records; diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index b90f1df9d7dae..ece9ae56ff725 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -405,10 +405,11 @@ private Map createFetchRequests(Cluster cluster) { fetchable.put(node, fetch); } - long offset = this.subscriptions.fetched(partition); + long fetched = this.subscriptions.fetched(partition); long consumed = this.subscriptions.consumed(partition); - if (consumed == offset) - fetch.put(partition, new FetchRequest.PartitionData(offset, this.fetchSize)); + // Only fetch data for partitions whose previously fetched data has been consumed + if (consumed == fetched) + fetch.put(partition, new FetchRequest.PartitionData(fetched, this.fetchSize)); } } From 2e8566855a6310706ae6c341987e5b056bf7fa1c Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Tue, 22 Sep 2015 16:16:33 +0000 Subject: [PATCH 17/18] add final and private keyword --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 2 +- .../apache/kafka/common/errors/OffsetOutOfRangeException.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 162264dc56776..91671cd802328 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -79,7 +79,7 @@ public class Fetcher { private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; - private Map offsetOutOfRangePartitions; + private final Map offsetOutOfRangePartitions; public Fetcher(ConsumerNetworkClient client, int minBytes, diff --git a/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java b/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java index 16f2e186d465a..4983bc0002ce0 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java @@ -22,7 +22,7 @@ public class OffsetOutOfRangeException extends RetriableException { private static final long serialVersionUID = 1L; - Map offsetOutOfRangePartitions = null; + private Map offsetOutOfRangePartitions = null; public OffsetOutOfRangeException() { } @@ -43,7 +43,7 @@ public OffsetOutOfRangeException(String message, Throwable cause) { super(message, cause); } - public Map getOutOfRangePartitions() { + public Map offsetOutOfRangePartitions() { return offsetOutOfRangePartitions; } From d71b1913e3895449b4c088d7666b48fb9a7182f7 Mon Sep 17 00:00:00 2001 From: Dong Lin Date: Wed, 23 Sep 2015 23:32:52 +0000 Subject: [PATCH 18/18] add comment --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 1 + 1 file changed, 1 insertion(+) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 91671cd802328..95480881503fd 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -278,6 +278,7 @@ private void throwIfOffsetOutOfRange() throws OffsetOutOfRangeException { continue; } Long consumed = subscriptions.consumed(entry.getKey()); + // ignore partition if its consumed offset != offset in fetchResponse, e.g. after seek() if (consumed != null && entry.getValue() == consumed) currentOutOfRangePartitions.put(entry.getKey(), entry.getValue()); }