From 95d1405f49c287475544ab8aead98533891ce159 Mon Sep 17 00:00:00 2001 From: Sean Glover Date: Wed, 12 Jun 2019 16:39:46 -0400 Subject: [PATCH 1/5] KafkaConsumer should not throw away already fetched data for paused partitions --- .../clients/consumer/internals/Fetcher.java | 118 +++++++++++++++--- .../consumer/internals/FetcherTest.java | 95 ++++++++++++++ 2 files changed, 195 insertions(+), 18 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 d4d028d38dc73..eb1625b2e5645 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,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; @@ -141,6 +143,7 @@ public class Fetcher implements Closeable { private final FetchManagerMetrics sensors; private final SubscriptionState subscriptions; private final ConcurrentLinkedQueue completedFetches; + private final ConcurrentLinkedQueue parsedFetchesCache; private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create(); private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; @@ -190,6 +193,7 @@ public Fetcher(LogContext logContext, this.keyDeserializer = keyDeserializer; this.valueDeserializer = valueDeserializer; this.completedFetches = new ConcurrentLinkedQueue<>(); + this.parsedFetchesCache = new ConcurrentLinkedQueue<>(); this.sensors = new FetchManagerMetrics(metrics, metricsRegistry); this.retryBackoffMs = retryBackoffMs; this.requestTimeoutMs = requestTimeoutMs; @@ -577,33 +581,42 @@ private Map beginningOrEndOffset(Collection>> fetchedRecords() { Map>> fetched = new HashMap<>(); + Queue pausedCompletedFetches = new ArrayDeque<>(); + Queue pausedParsedRecordsCache = 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(); + nextInLineRecords = maybeGetParsedFetchFromCache(pausedParsedRecordsCache); + + if (nextInLineRecords == null) { + CompletedFetch completedFetch = maybeGetCompletedFetch(pausedCompletedFetches); + + 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(); + } + throw e; } - throw e; + completedFetches.poll(); } - completedFetches.poll(); + } 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); @@ -623,10 +636,63 @@ public Map>> fetchedRecords() { } catch (KafkaException e) { if (fetched.isEmpty()) throw e; + } finally { + // add any partitions that were paused back to queues to be reevaluating in the next poll + completedFetches.addAll(pausedCompletedFetches); + parsedFetchesCache.addAll(pausedParsedRecordsCache); } + return fetched; } + /** + * Find the first non-paused parsed fetch of partition records. Parsed fetches of partition records are cached when + * they belong to a paused partition. They are cached so that they can be returned in the next poll when the + * partition may be resumed. Any parsed fetches that are still paused are added back to the cache after this poll + * for fetched records. + * @return A parsed fetch of partition records or null. + */ + private PartitionRecords maybeGetParsedFetchFromCache(Queue pausedParsedRecordsCache) { + PartitionRecords records = parsedFetchesCache.peek(); + if (records == null) return null; + + do { + if (subscriptions.isPaused(records.partition)) { + log.debug("Caching a previously parsed completed fetch for partition {} because it is still paused", records.partition); + pausedParsedRecordsCache.add(parsedFetchesCache.poll()); + records = parsedFetchesCache.peek(); + } else { + log.debug("A previously parsed completed fetch for partition {} is no longer paused and will be returned", records.partition); + parsedFetchesCache.poll(); + break; + } + } while (records != null); + + return records; + } + + /** + * Find the first completed fetch that belongs to a non-paused partition and return it. Record all paused completed + * fetches to be added back to the completed fetches queue after this poll for fetched records. + * @return A completed fetch or null. + */ + private CompletedFetch maybeGetCompletedFetch(Queue pausedCompletedFetches) { + CompletedFetch completedFetch = completedFetches.peek(); + if (completedFetch == null) return null; + + do { + if (subscriptions.isPaused(completedFetch.partition)) { + log.debug("Returning the completed fetch for partition {} to the back of the queue because its partitions are paused", completedFetch.partition); + pausedCompletedFetches.add(completedFetches.poll()); + completedFetch = completedFetches.peek(); + } else { + break; + } + } while (completedFetch != null); + + return completedFetch; + } + 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 +703,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 cache the records 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("Caching fetched records for assigned partition {} because it is paused", partitionRecords.partition); + parsedFetchesCache.add(partitionRecords); + nextInLineRecords = null; + return emptyList(); + } } else { SubscriptionState.FetchPosition position = subscriptions.position(partitionRecords.partition); if (partitionRecords.nextFetchOffset == position.offset) { @@ -661,6 +736,11 @@ private List> fetchRecords(PartitionRecords partitionRecord this.sensors.recordPartitionLead(partitionRecords.partition, lead); } + // TODO: is this necessary? are metrics still reported if we don't drain? + if (partitionRecords.lastRecord == null) { + partitionRecords.drain(); + } + return partRecords; } else { // these records aren't next in line based on the last consumed position, ignore them @@ -670,7 +750,9 @@ private List> fetchRecords(PartitionRecords partitionRecord } } + log.trace("Draining fetched records for partition {}", partitionRecords.partition); partitionRecords.drain(); + return emptyList(); } 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..319e930cfb891 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,101 @@ 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()); + 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()); + } + + @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()); + assertNotNull(fetchedRecords.get(tp1)); + assertNull(fetchedRecords.get(tp0)); + + fetchedRecords = fetchedRecords(); + assertEquals("Should return no records for remaining paused partition", 0, fetchedRecords.size()); + } + + @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()); + } + @Test public void testFetchNotLeaderForPartition() { buildFetcher(); From f042406a55e0dd87ae71e3f17c369074079ac49a Mon Sep 17 00:00:00 2001 From: Sean Glover Date: Thu, 4 Jul 2019 17:10:47 -0400 Subject: [PATCH 2/5] Exclude cached parsed records from fetchablePartitions, drain cached records when partitions no longer assigned. --- .../clients/consumer/internals/Fetcher.java | 36 +++++++++++++------ 1 file changed, 26 insertions(+), 10 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 eb1625b2e5645..7b55cc60adb97 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 @@ -227,6 +227,11 @@ public boolean hasCompletedFetches() { return !completedFetches.isEmpty(); } + // Visibilty for testing + protected boolean hasParsedFetchesCache() { + return !parsedFetchesCache.isEmpty(); + } + /** * 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. @@ -637,7 +642,7 @@ public Map>> fetchedRecords() { if (fetched.isEmpty()) throw e; } finally { - // add any partitions that were paused back to queues to be reevaluating in the next poll + // add any partitions that were paused back to queues to be re-evaluated in the next poll completedFetches.addAll(pausedCompletedFetches); parsedFetchesCache.addAll(pausedParsedRecordsCache); } @@ -736,11 +741,6 @@ private List> fetchRecords(PartitionRecords partitionRecord this.sensors.recordPartitionLead(partitionRecords.partition, lead); } - // TODO: is this necessary? are metrics still reported if we don't drain? - if (partitionRecords.lastRecord == null) { - partitionRecords.drain(); - } - return partRecords; } else { // these records aren't next in line based on the last consumed position, ignore them @@ -1110,6 +1110,9 @@ private List fetchablePartitions() { for (CompletedFetch completedFetch : completedFetches) { exclude.add(completedFetch.partition); } + for (PartitionRecords records : parsedFetchesCache) { + exclude.add(records.partition); + } return subscriptions.fetchablePartitions(tp -> !exclude.contains(tp)); } @@ -1368,19 +1371,32 @@ 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()) { + TopicPartition tp = completedFetchesItr.next().partition; if (!assignedPartitions.contains(tp)) { - itr.remove(); + completedFetchesItr.remove(); } } + + Iterator parsedFetchesCacheItr = parsedFetchesCache.iterator(); + while (parsedFetchesCacheItr.hasNext()) { + PartitionRecords records = parsedFetchesCacheItr.next(); + TopicPartition tp = records.partition; + if (!assignedPartitions.contains(tp)) { + records.drain(); + parsedFetchesCacheItr.remove(); + } + } + if (nextInLineRecords != null && !assignedPartitions.contains(nextInLineRecords.partition)) { nextInLineRecords.drain(); nextInLineRecords = null; } } + + /** * Clear the buffered data which are not a part of newly assigned topics * From ed60cfb241c8fd0ae78adb461612c00eac025ef5 Mon Sep 17 00:00:00 2001 From: Sean Glover Date: Thu, 4 Jul 2019 17:10:58 -0400 Subject: [PATCH 3/5] More tests --- .../consumer/internals/FetcherTest.java | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) 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 319e930cfb891..93e62454864ed 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 @@ -327,6 +327,7 @@ public void testClearBufferedDataForTopicPartitions() { fetcher.clearBufferedDataForUnassignedPartitions(newAssignedTopicPartitions); assertFalse(fetcher.hasCompletedFetches()); + assertFalse(fetcher.hasParsedFetchesCache()); } @Test @@ -913,6 +914,7 @@ public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() { 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()); @@ -927,6 +929,7 @@ public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() { 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 @@ -955,11 +958,13 @@ public void testFetchOnCompletedFetchesForSomePausedPartitions() { 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 @@ -990,6 +995,46 @@ public void testFetchOnCompletedFetchesForAllPausedPartitions() { 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 testPartialFetchOnParsedFetchCache() { + // 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 cached parsed fetches", fetcher.hasParsedFetchesCache()); + + 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 the parsed fetches cache", fetcher.hasParsedFetchesCache()); + + 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 cached parsed fetches", fetcher.hasParsedFetchesCache()); } @Test From 906d160886635ebbe969b7182fb66d81de5a345c Mon Sep 17 00:00:00 2001 From: Sean Glover Date: Fri, 26 Jul 2019 13:44:22 -0400 Subject: [PATCH 4/5] WIP- Move parseCompletedFetch to sendFetches callback --- .../clients/consumer/internals/Fetcher.java | 152 +++++------------- .../consumer/internals/FetcherTest.java | 9 +- 2 files changed, 48 insertions(+), 113 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 7b55cc60adb97..462d8a3a1b27b 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,6 @@ import java.io.Closeable; import java.nio.ByteBuffer; -import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -94,7 +93,6 @@ 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; @@ -142,8 +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 parsedFetchesCache; + private final ConcurrentLinkedQueue completedFetches; private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create(); private final Deserializer keyDeserializer; private final Deserializer valueDeserializer; @@ -193,7 +190,7 @@ public Fetcher(LogContext logContext, this.keyDeserializer = keyDeserializer; this.valueDeserializer = valueDeserializer; this.completedFetches = new ConcurrentLinkedQueue<>(); - this.parsedFetchesCache = new ConcurrentLinkedQueue<>(); + //this.parsedFetchesCache = new ConcurrentLinkedQueue<>(); this.sensors = new FetchManagerMetrics(metrics, metricsRegistry); this.retryBackoffMs = retryBackoffMs; this.requestTimeoutMs = requestTimeoutMs; @@ -227,11 +224,6 @@ public boolean hasCompletedFetches() { return !completedFetches.isEmpty(); } - // Visibilty for testing - protected boolean hasParsedFetchesCache() { - return !parsedFetchesCache.isEmpty(); - } - /** * 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. @@ -300,8 +292,26 @@ 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())); + // TODO: Do we even need `CompletedFetch` if we create a `PartitionRecords` right away? + CompletedFetch completedFetch = new CompletedFetch(partition, fetchOffset, + fetchData, metricAggregator, resp.requestHeader().apiVersion()); + + // TODO: Exception handling serves no purpose if we can't "cache" a `CompletedFetch` anywhere +// try { + PartitionRecords records = parseCompletedFetch(completedFetch); + if (records != null) + completedFetches.add(records); +// } catch (Exception e) { +// // Remove a completedFetch upon a parse with exception if it contains no records +// // This ensures that the completedFetches is not stuck with the same completedFetch +// // in cases such as the TopicAuthorizationException +// FetchResponse.PartitionData partitionData = completedFetch.partitionData; +// if (partitionData.records == null || partitionData.records.sizeInBytes() == 0) { +// completedFetches.poll(); +// } +// throw e; +// } + } } @@ -586,37 +596,14 @@ private Map beginningOrEndOffset(Collection>> fetchedRecords() { Map>> fetched = new HashMap<>(); - Queue pausedCompletedFetches = new ArrayDeque<>(); - Queue pausedParsedRecordsCache = new ArrayDeque<>(); int recordsRemaining = maxPollRecords; try { while (recordsRemaining > 0) { if (nextInLineRecords == null || nextInLineRecords.isFetched) { - nextInLineRecords = maybeGetParsedFetchFromCache(pausedParsedRecordsCache); - - if (nextInLineRecords == null) { - CompletedFetch completedFetch = maybeGetCompletedFetch(pausedCompletedFetches); - - 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(); - } - throw e; - } - completedFetches.poll(); - } + nextInLineRecords = maybeGetPartitionRecords(); + if (nextInLineRecords == null) break; } else { List> records = fetchRecords(nextInLineRecords, recordsRemaining); @@ -641,61 +628,24 @@ public Map>> fetchedRecords() { } catch (KafkaException e) { if (fetched.isEmpty()) throw e; - } finally { - // add any partitions that were paused back to queues to be re-evaluated in the next poll - completedFetches.addAll(pausedCompletedFetches); - parsedFetchesCache.addAll(pausedParsedRecordsCache); } return fetched; } - /** - * Find the first non-paused parsed fetch of partition records. Parsed fetches of partition records are cached when - * they belong to a paused partition. They are cached so that they can be returned in the next poll when the - * partition may be resumed. Any parsed fetches that are still paused are added back to the cache after this poll - * for fetched records. - * @return A parsed fetch of partition records or null. - */ - private PartitionRecords maybeGetParsedFetchFromCache(Queue pausedParsedRecordsCache) { - PartitionRecords records = parsedFetchesCache.peek(); - if (records == null) return null; - - do { + private PartitionRecords maybeGetPartitionRecords() { + PartitionRecords partRecords = null; + for (Iterator it = completedFetches.iterator(); it.hasNext(); ) { + PartitionRecords records = it.next(); if (subscriptions.isPaused(records.partition)) { - log.debug("Caching a previously parsed completed fetch for partition {} because it is still paused", records.partition); - pausedParsedRecordsCache.add(parsedFetchesCache.poll()); - records = parsedFetchesCache.peek(); + log.debug("Skipping the completed fetch for partition {} because its partition is paused", records.partition); } else { - log.debug("A previously parsed completed fetch for partition {} is no longer paused and will be returned", records.partition); - parsedFetchesCache.poll(); + partRecords = records; + it.remove(); break; } - } while (records != null); - - return records; - } - - /** - * Find the first completed fetch that belongs to a non-paused partition and return it. Record all paused completed - * fetches to be added back to the completed fetches queue after this poll for fetched records. - * @return A completed fetch or null. - */ - private CompletedFetch maybeGetCompletedFetch(Queue pausedCompletedFetches) { - CompletedFetch completedFetch = completedFetches.peek(); - if (completedFetch == null) return null; - - do { - if (subscriptions.isPaused(completedFetch.partition)) { - log.debug("Returning the completed fetch for partition {} to the back of the queue because its partitions are paused", completedFetch.partition); - pausedCompletedFetches.add(completedFetches.poll()); - completedFetch = completedFetches.peek(); - } else { - break; - } - } while (completedFetch != null); - - return completedFetch; + } + return partRecords; } private List> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { @@ -709,11 +659,11 @@ private List> fetchRecords(PartitionRecords partitionRecord log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); - // when the partition is paused we cache the records instead of draining them so that they can be returned - // on a subsequent poll if the partition is resumed + // 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("Caching fetched records for assigned partition {} because it is paused", partitionRecords.partition); - parsedFetchesCache.add(partitionRecords); + log.debug("Skipping fetching records for assigned partition {} because it is paused", partitionRecords.partition); + completedFetches.add(partitionRecords); nextInLineRecords = null; return emptyList(); } @@ -1107,12 +1057,9 @@ private List fetchablePartitions() { if (nextInLineRecords != null && !nextInLineRecords.isFetched) { exclude.add(nextInLineRecords.partition); } - for (CompletedFetch completedFetch : completedFetches) { + for (PartitionRecords completedFetch : completedFetches) { exclude.add(completedFetch.partition); } - for (PartitionRecords records : parsedFetchesCache) { - exclude.add(records.partition); - } return subscriptions.fetchablePartitions(tp -> !exclude.contains(tp)); } @@ -1220,10 +1167,9 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) { 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 @@ -1371,21 +1317,13 @@ private Optional maybeLeaderEpoch(int leaderEpoch) { * @param assignedPartitions newly assigned {@link TopicPartition} */ public void clearBufferedDataForUnassignedPartitions(Collection assignedPartitions) { - Iterator completedFetchesItr = completedFetches.iterator(); + Iterator completedFetchesItr = completedFetches.iterator(); while (completedFetchesItr.hasNext()) { - TopicPartition tp = completedFetchesItr.next().partition; - if (!assignedPartitions.contains(tp)) { - completedFetchesItr.remove(); - } - } - - Iterator parsedFetchesCacheItr = parsedFetchesCache.iterator(); - while (parsedFetchesCacheItr.hasNext()) { - PartitionRecords records = parsedFetchesCacheItr.next(); + PartitionRecords records = completedFetchesItr.next(); TopicPartition tp = records.partition; if (!assignedPartitions.contains(tp)) { records.drain(); - parsedFetchesCacheItr.remove(); + completedFetchesItr.remove(); } } @@ -1395,8 +1333,6 @@ public void clearBufferedDataForUnassignedPartitions(Collection } } - - /** * Clear the buffered data which are not a part of newly assigned topics * 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 93e62454864ed..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 @@ -327,7 +327,6 @@ public void testClearBufferedDataForTopicPartitions() { fetcher.clearBufferedDataForUnassignedPartitions(newAssignedTopicPartitions); assertFalse(fetcher.hasCompletedFetches()); - assertFalse(fetcher.hasParsedFetchesCache()); } @Test @@ -999,7 +998,7 @@ public void testFetchOnCompletedFetchesForAllPausedPartitions() { } @Test - public void testPartialFetchOnParsedFetchCache() { + 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 @@ -1017,7 +1016,7 @@ public void testPartialFetchOnParsedFetchCache() { fetchedRecords = fetchedRecords(); assertEquals("Should return 2 records from fetch with 3 records", 2, fetchedRecords.get(tp0).size()); - assertFalse("Should have no cached parsed fetches", fetcher.hasParsedFetchesCache()); + assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches()); subscriptions.pause(tp0); consumerClient.poll(time.timer(0)); @@ -1025,7 +1024,7 @@ public void testPartialFetchOnParsedFetchCache() { fetchedRecords = fetchedRecords(); assertEquals("Should return no records for paused partitions", 0, fetchedRecords.size()); - assertTrue("Should have 1 entry in the parsed fetches cache", fetcher.hasParsedFetchesCache()); + assertTrue("Should have 1 entry in completed fetches", fetcher.hasCompletedFetches()); subscriptions.resume(tp0); @@ -1034,7 +1033,7 @@ public void testPartialFetchOnParsedFetchCache() { fetchedRecords = fetchedRecords(); assertEquals("Should return last remaining record", 1, fetchedRecords.get(tp0).size()); - assertFalse("Should have no cached parsed fetches", fetcher.hasParsedFetchesCache()); + assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches()); } @Test From a4e2532aca4b2dd4cbc286cec894633d4b652508 Mon Sep 17 00:00:00 2001 From: Sean Glover Date: Sun, 28 Jul 2019 14:55:24 -0400 Subject: [PATCH 5/5] Initialize PartitionRecords in fetch callback --- .../clients/consumer/internals/Fetcher.java | 73 ++++++++++++------- 1 file changed, 47 insertions(+), 26 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 462d8a3a1b27b..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 @@ -190,7 +190,6 @@ public Fetcher(LogContext logContext, this.keyDeserializer = keyDeserializer; this.valueDeserializer = valueDeserializer; this.completedFetches = new ConcurrentLinkedQueue<>(); - //this.parsedFetchesCache = new ConcurrentLinkedQueue<>(); this.sensors = new FetchManagerMetrics(metrics, metricsRegistry); this.retryBackoffMs = retryBackoffMs; this.requestTimeoutMs = requestTimeoutMs; @@ -292,26 +291,11 @@ public void onSuccess(ClientResponse resp) { log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", isolationLevel, fetchOffset, partition, fetchData); - // TODO: Do we even need `CompletedFetch` if we create a `PartitionRecords` right away? + CompletedFetch completedFetch = new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator, resp.requestHeader().apiVersion()); - // TODO: Exception handling serves no purpose if we can't "cache" a `CompletedFetch` anywhere -// try { - PartitionRecords records = parseCompletedFetch(completedFetch); - if (records != null) - completedFetches.add(records); -// } catch (Exception e) { -// // Remove a completedFetch upon a parse with exception if it contains no records -// // This ensures that the completedFetches is not stuck with the same completedFetch -// // in cases such as the TopicAuthorizationException -// FetchResponse.PartitionData partitionData = completedFetch.partitionData; -// if (partitionData.records == null || partitionData.records.sizeInBytes() == 0) { -// completedFetches.poll(); -// } -// throw e; -// } - + completedFetches.add(parseCompletedFetch(completedFetch)); } } @@ -601,9 +585,29 @@ public Map>> fetchedRecords() { try { while (recordsRemaining > 0) { if (nextInLineRecords == null || nextInLineRecords.isFetched) { - nextInLineRecords = maybeGetPartitionRecords(); - - if (nextInLineRecords == null) break; + 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; + } + } else { + nextInLineRecords = records; + } + completedFetches.remove(records); } else { List> records = fetchRecords(nextInLineRecords, recordsRemaining); @@ -635,13 +639,11 @@ public Map>> fetchedRecords() { private PartitionRecords maybeGetPartitionRecords() { PartitionRecords partRecords = null; - for (Iterator it = completedFetches.iterator(); it.hasNext(); ) { - PartitionRecords records = it.next(); + 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; - it.remove(); break; } } @@ -1157,11 +1159,23 @@ 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; @@ -1183,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) { @@ -1227,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 || @@ -1379,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, @@ -1580,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 {