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 30944b330da60..b9e60ee364c18 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 @@ -1269,7 +1269,7 @@ private Map>> pollForFetches(Timer tim client.poll(pollTimer, () -> { // since a fetch might be completed by the background thread, we need this poll condition // to ensure that we do not block unnecessarily in poll() - return !fetcher.hasCompletedFetches(); + return !fetcher.hasAvailableFetches(); }); timer.update(pollTimer.currentTimeMs()); 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 d4d028d38dc73..c1c42bd06ca4f 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 @@ -50,8 +50,8 @@ import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Meter; import org.apache.kafka.common.metrics.stats.Min; -import org.apache.kafka.common.metrics.stats.WindowedCount; import org.apache.kafka.common.metrics.stats.Value; +import org.apache.kafka.common.metrics.stats.WindowedCount; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.BufferSupplier; @@ -81,6 +81,7 @@ import java.io.Closeable; import java.nio.ByteBuffer; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -93,6 +94,7 @@ import java.util.Map; import java.util.Optional; import java.util.PriorityQueue; +import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; @@ -140,7 +142,7 @@ public class Fetcher implements Closeable { private final ConsumerMetadata metadata; private final FetchManagerMetrics sensors; private final SubscriptionState subscriptions; - private final ConcurrentLinkedQueue completedFetches; + private final ConcurrentLinkedQueue completedFetches; private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create(); private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; @@ -216,13 +218,22 @@ private static class ListOffsetData { } /** - * Return whether we have any completed fetches pending return to the user. This method is thread-safe. + * Return whether we have any completed fetches pending return to the user. This method is thread-safe. Has + * visibility for testing. * @return true if there are completed fetches, false otherwise */ - public boolean hasCompletedFetches() { + protected boolean hasCompletedFetches() { return !completedFetches.isEmpty(); } + /** + * Return whether we have any completed fetches that are fetchable. This method is thread-safe. + * @return true if there are completed fetches that can be returned, false otherwise + */ + public boolean hasAvailableFetches() { + return completedFetches.stream().anyMatch(fetch -> subscriptions.isFetchable(fetch.partition)); + } + /** * Set-up a fetch request for any node that we have assigned partitions for which doesn't already have * an in-flight fetch or pending fetch data. @@ -291,8 +302,11 @@ public void onSuccess(ClientResponse resp) { log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", isolationLevel, fetchOffset, partition, fetchData); - completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator, - resp.requestHeader().apiVersion())); + + CompletedFetch completedFetch = new CompletedFetch(partition, fetchOffset, + fetchData, metricAggregator, resp.requestHeader().apiVersion()); + + completedFetches.add(parseCompletedFetch(completedFetch)); } } @@ -577,33 +591,45 @@ private Map beginningOrEndOffset(Collection>> fetchedRecords() { Map>> fetched = new HashMap<>(); + Queue pausedCompletedFetches = new ArrayDeque<>(); int recordsRemaining = maxPollRecords; try { while (recordsRemaining > 0) { if (nextInLineRecords == null || nextInLineRecords.isFetched) { - CompletedFetch completedFetch = completedFetches.peek(); - if (completedFetch == null) break; - - try { - nextInLineRecords = parseCompletedFetch(completedFetch); - } catch (Exception e) { - // Remove a completedFetch upon a parse with exception if (1) it contains no records, and - // (2) there are no fetched records with actual content preceding this exception. - // The first condition ensures that the completedFetches is not stuck with the same completedFetch - // in cases such as the TopicAuthorizationException, and the second condition ensures that no - // potential data loss due to an exception in a following record. - FetchResponse.PartitionData partition = completedFetch.partitionData; - if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) { - completedFetches.poll(); + PartitionRecords records = completedFetches.peek(); + if (records == null) break; + + if (records.notInitialized()) { + try { + nextInLineRecords = initializePartitionRecords(records); + } catch (Exception e) { + // Remove a completedFetch upon a parse with exception if (1) it contains no records, and + // (2) there are no fetched records with actual content preceding this exception. + // The first condition ensures that the completedFetches is not stuck with the same completedFetch + // in cases such as the TopicAuthorizationException, and the second condition ensures that no + // potential data loss due to an exception in a following record. + FetchResponse.PartitionData partition = records.completedFetch.partitionData; + if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) { + completedFetches.poll(); + } + throw e; } - throw e; + } else { + nextInLineRecords = records; } completedFetches.poll(); + } else if (subscriptions.isPaused(nextInLineRecords.partition)) { + // when the partition is paused we add the records back to the completedFetches queue instead of draining + // them so that they can be returned on a subsequent poll if the partition is resumed at that time + log.debug("Skipping fetching records for assigned partition {} because it is paused", nextInLineRecords.partition); + pausedCompletedFetches.add(nextInLineRecords); + nextInLineRecords = null; } else { List> records = fetchRecords(nextInLineRecords, recordsRemaining); - TopicPartition partition = nextInLineRecords.partition; + if (!records.isEmpty()) { + TopicPartition partition = nextInLineRecords.partition; List> currentRecords = fetched.get(partition); if (currentRecords == null) { fetched.put(partition, records); @@ -623,7 +649,12 @@ public Map>> fetchedRecords() { } catch (KafkaException e) { if (fetched.isEmpty()) throw e; + } finally { + // add any polled completed fetches for paused partitions back to the completed fetches queue to be + // re-evaluated in the next poll + completedFetches.addAll(pausedCompletedFetches); } + return fetched; } @@ -670,7 +701,9 @@ private List> fetchRecords(PartitionRecords partitionRecord } } + log.trace("Draining fetched records for partition {}", partitionRecords.partition); partitionRecords.drain(); + return emptyList(); } @@ -1025,7 +1058,7 @@ private List fetchablePartitions() { if (nextInLineRecords != null && !nextInLineRecords.isFetched) { exclude.add(nextInLineRecords.partition); } - for (CompletedFetch completedFetch : completedFetches) { + for (PartitionRecords completedFetch : completedFetches) { exclude.add(completedFetch.partition); } return subscriptions.fetchablePartitions(tp -> !exclude.contains(tp)); @@ -1125,20 +1158,31 @@ private Map> regroupPartitionMapByNode(Map partition = completedFetch.partitionData; + + Iterator batches = partition.records.batches().iterator(); + return new PartitionRecords(tp, completedFetch, batches); + } + + /** + * Initialize a PartitionRecords object. + */ + private PartitionRecords initializePartitionRecords(PartitionRecords partitionRecordsToInitialize) { + CompletedFetch completedFetch = partitionRecordsToInitialize.completedFetch; + TopicPartition tp = completedFetch.partition; + FetchResponse.PartitionData partition = completedFetch.partitionData; long fetchOffset = completedFetch.fetchedOffset; PartitionRecords partitionRecords = null; Errors error = partition.error; try { - if (!subscriptions.isFetchable(tp)) { - // this can happen when a rebalance happened or a partition consumption paused - // while fetch is still in-flight - log.debug("Ignoring fetched records for partition {} since it is no longer fetchable", tp); + if (!subscriptions.hasValidPosition(tp)) { + // this can happen when a rebalance happened while fetch is still in-flight + log.debug("Ignoring fetched records for partition {} since it no longer has valid position", tp); } else if (error == Errors.NONE) { // we are interested in this fetch only if the beginning offset matches the // current consumed position @@ -1152,7 +1196,7 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) { log.trace("Preparing to read {} bytes of data for partition {} with offset {}", partition.records.sizeInBytes(), tp, position); Iterator batches = partition.records.batches().iterator(); - partitionRecords = new PartitionRecords(tp, completedFetch, batches); + partitionRecords = partitionRecordsToInitialize; if (!batches.hasNext() && partition.records.sizeInBytes() > 0) { if (completedFetch.responseVersion < 3) { @@ -1196,6 +1240,8 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) { }); } + + partitionRecordsToInitialize.initialized = true; } else if (error == Errors.NOT_LEADER_FOR_PARTITION || error == Errors.REPLICA_NOT_AVAILABLE || error == Errors.KAFKA_STORAGE_ERROR || @@ -1286,13 +1332,16 @@ private Optional maybeLeaderEpoch(int leaderEpoch) { * @param assignedPartitions newly assigned {@link TopicPartition} */ public void clearBufferedDataForUnassignedPartitions(Collection assignedPartitions) { - Iterator itr = completedFetches.iterator(); - while (itr.hasNext()) { - TopicPartition tp = itr.next().partition; + Iterator completedFetchesItr = completedFetches.iterator(); + while (completedFetchesItr.hasNext()) { + PartitionRecords records = completedFetchesItr.next(); + TopicPartition tp = records.partition; if (!assignedPartitions.contains(tp)) { - itr.remove(); + records.drain(); + completedFetchesItr.remove(); } } + if (nextInLineRecords != null && !assignedPartitions.contains(nextInLineRecords.partition)) { nextInLineRecords.drain(); nextInLineRecords = null; @@ -1345,6 +1394,7 @@ private class PartitionRecords { private boolean isFetched = false; private Exception cachedRecordException = null; private boolean corruptLastRecord = false; + private boolean initialized = false; private PartitionRecords(TopicPartition partition, CompletedFetch completedFetch, @@ -1546,6 +1596,10 @@ private boolean containsAbortMarker(RecordBatch batch) { Record firstRecord = batchIterator.next(); return ControlRecordType.ABORT == ControlRecordType.parse(firstRecord.key()); } + + private boolean notInitialized() { + return !this.initialized; + } } private static class CompletedFetch { 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 44c00c48a3ee6..5f66689520f36 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 @@ -897,6 +897,171 @@ public void testFetchOnPausedPartition() { assertTrue(client.requests().isEmpty()); } + @Test + public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() { + buildFetcher(); + + assignFromUser(singleton(tp0)); + subscriptions.seek(tp0, 0); + + assertEquals(1, fetcher.sendFetches()); + + subscriptions.pause(tp0); + + client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); + consumerClient.poll(time.timer(0)); + + Map>> fetchedRecords = fetchedRecords(); + assertEquals("Should not return any records when partition is paused", 0, fetchedRecords.size()); + assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches()); + assertFalse("Should not have any available (non-paused) completed fetches", fetcher.hasAvailableFetches()); + assertNull(fetchedRecords.get(tp0)); + assertEquals(0, fetcher.sendFetches()); + + subscriptions.resume(tp0); + + assertTrue("Should have available (non-paused) completed fetches", fetcher.hasAvailableFetches()); + + consumerClient.poll(time.timer(0)); + fetchedRecords = fetchedRecords(); + assertEquals("Should return records when partition is resumed", 1, fetchedRecords.size()); + assertNotNull(fetchedRecords.get(tp0)); + assertEquals(3, fetchedRecords.get(tp0).size()); + + consumerClient.poll(time.timer(0)); + fetchedRecords = fetchedRecords(); + assertEquals("Should not return records after previously paused partitions are fetched", 0, fetchedRecords.size()); + assertFalse("Should no longer contain completed fetches", fetcher.hasCompletedFetches()); + } + + @Test + public void testFetchOnCompletedFetchesForSomePausedPartitions() { + buildFetcher(); + + Map>> fetchedRecords; + + assignFromUser(Utils.mkSet(tp0, tp1)); + + // seek to tp0 and tp1 in two polls to generate 2 complete requests and responses + + // #1 seek, request, poll, response + subscriptions.seek(tp0, 1); + assertEquals(1, fetcher.sendFetches()); + client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); + consumerClient.poll(time.timer(0)); + + // #2 seek, request, poll, response + subscriptions.seek(tp1, 1); + assertEquals(1, fetcher.sendFetches()); + client.prepareResponse(fullFetchResponse(tp1, this.nextRecords, Errors.NONE, 100L, 0)); + + subscriptions.pause(tp0); + consumerClient.poll(time.timer(0)); + + fetchedRecords = fetchedRecords(); + assertEquals("Should return completed fetch for unpaused partitions", 1, fetchedRecords.size()); + assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches()); + assertNotNull(fetchedRecords.get(tp1)); + assertNull(fetchedRecords.get(tp0)); + + fetchedRecords = fetchedRecords(); + assertEquals("Should return no records for remaining paused partition", 0, fetchedRecords.size()); + assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches()); + } + + @Test + public void testFetchOnCompletedFetchesForAllPausedPartitions() { + buildFetcher(); + + Map>> fetchedRecords; + + assignFromUser(Utils.mkSet(tp0, tp1)); + + // seek to tp0 and tp1 in two polls to generate 2 complete requests and responses + + // #1 seek, request, poll, response + subscriptions.seek(tp0, 1); + assertEquals(1, fetcher.sendFetches()); + client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); + consumerClient.poll(time.timer(0)); + + // #2 seek, request, poll, response + subscriptions.seek(tp1, 1); + assertEquals(1, fetcher.sendFetches()); + client.prepareResponse(fullFetchResponse(tp1, this.nextRecords, Errors.NONE, 100L, 0)); + + subscriptions.pause(tp0); + subscriptions.pause(tp1); + + consumerClient.poll(time.timer(0)); + + fetchedRecords = fetchedRecords(); + assertEquals("Should return no records for all paused partitions", 0, fetchedRecords.size()); + assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches()); + assertFalse("Should not have any available (non-paused) completed fetches", fetcher.hasAvailableFetches()); + } + + @Test + public void testPartialFetchWithPausedPartitions() { + // this test sends creates a completed fetch with 3 records and a max poll of 2 records to assert + // that a fetch that must be returned over at least 2 polls can be cached successfully when its partition is + // paused, then returned successfully after its been resumed again later + buildFetcher(2); + + Map>> fetchedRecords; + + assignFromUser(Utils.mkSet(tp0, tp1)); + + subscriptions.seek(tp0, 1); + assertEquals(1, fetcher.sendFetches()); + client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); + consumerClient.poll(time.timer(0)); + + fetchedRecords = fetchedRecords(); + + assertEquals("Should return 2 records from fetch with 3 records", 2, fetchedRecords.get(tp0).size()); + assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches()); + + subscriptions.pause(tp0); + consumerClient.poll(time.timer(0)); + + fetchedRecords = fetchedRecords(); + + assertEquals("Should return no records for paused partitions", 0, fetchedRecords.size()); + assertTrue("Should have 1 entry in completed fetches", fetcher.hasCompletedFetches()); + assertFalse("Should not have any available (non-paused) completed fetches", fetcher.hasAvailableFetches()); + + subscriptions.resume(tp0); + + consumerClient.poll(time.timer(0)); + + fetchedRecords = fetchedRecords(); + + assertEquals("Should return last remaining record", 1, fetchedRecords.get(tp0).size()); + assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches()); + } + + @Test + public void testFetchDiscardedAfterPausedPartitionResumedAndSeekedToNewOffset() { + buildFetcher(); + assignFromUser(singleton(tp0)); + subscriptions.seek(tp0, 0); + + assertEquals(1, fetcher.sendFetches()); + subscriptions.pause(tp0); + client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); + + subscriptions.seek(tp0, 3); + subscriptions.resume(tp0); + consumerClient.poll(time.timer(0)); + + assertTrue("Should have 1 entry in completed fetches", fetcher.hasCompletedFetches()); + Map>> fetchedRecords = fetchedRecords(); + assertEquals("Should not return any records because we seeked to a new offset", 0, fetchedRecords.size()); + assertNull(fetchedRecords.get(tp0)); + assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches()); + } + @Test public void testFetchNotLeaderForPartition() { buildFetcher();