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 all 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
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1269,7 +1269,7 @@ private Map<TopicPartition, List<ConsumerRecord<K, V>>> 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());

Expand Down
120 changes: 87 additions & 33 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 @@ -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.
Expand Down Expand Up @@ -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));
}
}

Expand Down Expand Up @@ -577,33 +591,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 +649,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 +701,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 +1058,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 +1158,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 +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<? 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 +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 ||
Expand Down Expand Up @@ -1286,13 +1332,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 +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,
Expand Down Expand Up @@ -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 {
Expand Down
Loading