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..43b692da2fd66 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 @@ -140,7 +140,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; @@ -291,8 +291,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)); } } @@ -582,28 +585,34 @@ public Map>> fetchedRecords() { 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 = maybeGetPartitionRecords(); + + 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.remove(records); + } + throw e; } - throw e; + } else { + nextInLineRecords = records; } - completedFetches.poll(); + completedFetches.remove(records); } else { List> records = fetchRecords(nextInLineRecords, recordsRemaining); - TopicPartition partition = nextInLineRecords.partition; - if (!records.isEmpty()) { + + if (!records.isEmpty() && nextInLineRecords != null) { + TopicPartition partition = nextInLineRecords.partition; List> currentRecords = fetched.get(partition); if (currentRecords == null) { fetched.put(partition, records); @@ -624,9 +633,23 @@ public Map>> fetchedRecords() { if (fetched.isEmpty()) throw e; } + return fetched; } + private PartitionRecords maybeGetPartitionRecords() { + PartitionRecords partRecords = null; + for (PartitionRecords records : completedFetches) { + if (subscriptions.isPaused(records.partition)) { + log.debug("Skipping the completed fetch for partition {} because its partition is paused", records.partition); + } else { + partRecords = records; + break; + } + } + return partRecords; + } + private List> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { // this can happen when a rebalance happened before fetched records are returned to the consumer's poll call @@ -637,6 +660,15 @@ private List> fetchRecords(PartitionRecords partitionRecord // poll call or if the offset is being reset log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); + + // when the partition is paused we add them back to completedFetches instead of draining them so that they + // can be returned on a subsequent poll if the partition is resumed + if (subscriptions.isPaused(partitionRecords.partition)) { + log.debug("Skipping fetching records for assigned partition {} because it is paused", partitionRecords.partition); + completedFetches.add(partitionRecords); + nextInLineRecords = null; + return emptyList(); + } } else { SubscriptionState.FetchPosition position = subscriptions.position(partitionRecords.partition); if (partitionRecords.nextFetchOffset == position.offset) { @@ -670,7 +702,9 @@ private List> fetchRecords(PartitionRecords partitionRecord } } + log.trace("Draining fetched records for partition {}", partitionRecords.partition); partitionRecords.drain(); + return emptyList(); } @@ -1025,7 +1059,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 +1159,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 +1197,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 +1241,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 +1333,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 +1395,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 +1597,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..b07f663369cd1 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,145 @@ 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()); + assertNull(fetchedRecords.get(tp0)); + assertEquals(0, fetcher.sendFetches()); + + subscriptions.resume(tp0); + + 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()); + } + + @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()); + + 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 testFetchNotLeaderForPartition() { buildFetcher();