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 3ac2be840c5d3..e1cfd7ed96fe7 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 @@ -814,6 +814,12 @@ private Map>> pollOnce(long timeout) { // init any new fetches (won't resend pending fetches) Cluster cluster = this.metadata.fetch(); + 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; + } 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 0efd34d1c1294..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 @@ -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; @@ -78,6 +79,8 @@ public class Fetcher { private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; + private final Map offsetOutOfRangePartitions; + public Fetcher(ConsumerNetworkClient client, int minBytes, int maxWaitMs, @@ -106,6 +109,7 @@ public Fetcher(ConsumerNetworkClient client, this.valueDeserializer = valueDeserializer; this.records = new LinkedList>(); + this.offsetOutOfRangePartitions = new HashMap<>(); this.sensors = new FetchManagerMetrics(metrics, metricGrpPrefix, metricTags); this.retryBackoffMs = retryBackoffMs; @@ -137,6 +141,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 @@ -257,16 +262,44 @@ 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()); + // 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()); + } + this.offsetOutOfRangePartitions.clear(); + if (!currentOutOfRangePartitions.isEmpty()) + throw new OffsetOutOfRangeException(currentOutOfRangePartitions); + } + /** * 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() { if (this.subscriptions.partitionAssignmentNeeded()) { return Collections.emptyMap(); } else { Map>> drained = new HashMap<>(); + throwIfOffsetOutOfRange(); + for (PartitionRecords part : this.records) { if (!subscriptions.isFetchable(part.partition)) { log.debug("Ignoring fetched records for {} since it is no longer fetchable", part.partition); @@ -378,8 +411,11 @@ private Map createFetchRequests(Cluster cluster) { fetchable.put(node, fetch); } - long offset = this.subscriptions.fetched(partition); - fetch.put(partition, new FetchRequest.PartitionData(offset, this.fetchSize)); + long fetched = this.subscriptions.fetched(partition); + long consumed = this.subscriptions.consumed(partition); + // Only fetch data for partitions whose previously fetched data has been consumed + if (consumed == fetched) + fetch.put(partition, new FetchRequest.PartitionData(fetched, this.fetchSize)); } } @@ -447,9 +483,12 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) { || 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 + 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)); - 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 c92a58119af82..225d2cb5b6919 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 @@ -244,6 +244,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/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java b/clients/src/main/java/org/apache/kafka/common/errors/OffsetOutOfRangeException.java index fc7c6e3471b05..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 @@ -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; + private 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 offsetOutOfRangePartitions() { + return offsetOutOfRangePartitions; + } + }