Skip to content

KAFKA-7548: KafkaConsumer should not throw away already fetched data for paused partitions (v2) #6988

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 8 commits into from
Aug 2, 2019
Merged

KAFKA-7548: KafkaConsumer should not throw away already fetched data for paused partitions (v2) #6988

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8f832e0
KafkaConsumer should not throw away already fetched data for paused p…
seglo Jun 12, 2019
fd57ff3
Exclude cached parsed records from fetchablePartitions, drain cached …
seglo Jul 4, 2019
fbefb25
More tests
seglo Jul 4, 2019
9095cda
Parse PartitionRecords in sendFetches callback and use completedFetch…
seglo Jul 26, 2019
ba68212
Cache paused partitions within fetchedRecords
seglo Jul 29, 2019
ded3f61
Simplify paused partition checking
hachikuji Jul 31, 2019
2f64a00
Implement hasAvailablePartitions for KafkaConsumer long poll
seglo Aug 1, 2019
c9b4719
New test to confirm completed fetch discarded when seeked to new offset
seglo Aug 1, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 76 additions & 31 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -140,7 +142,7 @@ public class Fetcher<K, V> implements Closeable {
private final ConsumerMetadata metadata;
private final FetchManagerMetrics sensors;
private final SubscriptionState subscriptions;
private final ConcurrentLinkedQueue<CompletedFetch> completedFetches;
private final ConcurrentLinkedQueue<PartitionRecords> completedFetches;
private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create();
private final Deserializer<K> keyDeserializer;
private final Deserializer<V> valueDeserializer;
Expand Down Expand Up @@ -291,8 +293,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));
}
}

Expand Down Expand Up @@ -577,33 +582,45 @@ private Map<TopicPartition, Long> beginningOrEndOffset(Collection<TopicPartition
*/
public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
Comment thread
hachikuji marked this conversation as resolved.
Map<TopicPartition, List<ConsumerRecord<K, V>>> fetched = new HashMap<>();
Queue<PartitionRecords> 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<ConsumerRecord<K, V>> records = fetchRecords(nextInLineRecords, recordsRemaining);
TopicPartition partition = nextInLineRecords.partition;

if (!records.isEmpty()) {
TopicPartition partition = nextInLineRecords.partition;
List<ConsumerRecord<K, V>> currentRecords = fetched.get(partition);
if (currentRecords == null) {
fetched.put(partition, records);
Expand All @@ -623,7 +640,12 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> 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;
}

Expand Down Expand Up @@ -670,7 +692,9 @@ private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecord
}
}

log.trace("Draining fetched records for partition {}", partitionRecords.partition);
partitionRecords.drain();

return emptyList();
}

Expand Down Expand Up @@ -1025,7 +1049,7 @@ private List<TopicPartition> 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));
Expand Down Expand Up @@ -1125,20 +1149,31 @@ private <T> Map<Node, Map<TopicPartition, T>> regroupPartitionMapByNode(Map<Topi
}

/**
* The callback for fetch completion
* Parse a PartitionRecords object from a CompletedFetch
*/
private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
TopicPartition tp = completedFetch.partition;
FetchResponse.PartitionData<Records> partition = completedFetch.partitionData;

Iterator<? extends RecordBatch> 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<Records> 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
Expand All @@ -1152,7 +1187,7 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
log.trace("Preparing to read {} bytes of data for partition {} with offset {}",
partition.records.sizeInBytes(), tp, position);
Iterator<? extends RecordBatch> batches = partition.records.batches().iterator();
partitionRecords = new PartitionRecords(tp, completedFetch, batches);
partitionRecords = partitionRecordsToInitialize;

if (!batches.hasNext() && partition.records.sizeInBytes() > 0) {
if (completedFetch.responseVersion < 3) {
Expand Down Expand Up @@ -1196,6 +1231,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 ||
Expand Down Expand Up @@ -1286,13 +1323,16 @@ private Optional<Integer> maybeLeaderEpoch(int leaderEpoch) {
* @param assignedPartitions newly assigned {@link TopicPartition}
*/
public void clearBufferedDataForUnassignedPartitions(Collection<TopicPartition> assignedPartitions) {
Iterator<CompletedFetch> itr = completedFetches.iterator();
while (itr.hasNext()) {
TopicPartition tp = itr.next().partition;
Iterator<PartitionRecords> 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;
Expand Down Expand Up @@ -1345,6 +1385,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,
Expand Down Expand Up @@ -1546,6 +1587,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 {
Expand Down
139 changes: 139 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -897,6 +897,145 @@ public void testFetchOnPausedPartition() {
assertTrue(client.requests().isEmpty());
}

@Test

@hachikuji hachikuji Aug 1, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a test case which covers the case where the user seeks to a new offset while a partition is paused with data available to return? In this case, we expect the data to be discarded when the partition is resumed.

@seglo seglo Aug 1, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did a pass over FetcherTest and didn't see this scenario exactly. I added another test called testFetchDiscardedAfterPausedPartitionResumedAndSeekedToNewOffset (a bit of a mouthful). Does it create the scenario you were thinking?

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<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> 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<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> 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<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> 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<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> 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();
Expand Down