Skip to content

KAFKA-7548 : KafkaConsumer should not throw away already fetched data for paused partitions. #5844

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

Closed
MayureshGharat wants to merge 1 commit into from
Closed

KAFKA-7548 : KafkaConsumer should not throw away already fetched data for paused partitions. #5844

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 4 additions & 0 deletions 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 @@ -1823,6 +1823,7 @@ public void pause(Collection<TopicPartition> partitions) {
try {
log.debug("Pausing partitions {}", partitions);
for (TopicPartition partition: partitions) {
fetcher.removeRecentlyUnpausedPartition(partition);
subscriptions.pause(partition);
}
} finally {
Expand All @@ -1843,6 +1844,9 @@ public void resume(Collection<TopicPartition> partitions) {
try {
log.debug("Resuming partitions {}", partitions);
for (TopicPartition partition: partitions) {
if (subscriptions.isPaused(partition)) {
fetcher.addRecentlyUnpausedPartition(partition);
}
subscriptions.resume(partition);
}
} finally {
Expand Down
139 changes: 121 additions & 18 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 @@ -71,6 +71,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 @@ -79,10 +80,12 @@
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
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 @@ -130,6 +133,9 @@ public class Fetcher<K, V> implements SubscriptionState.Listener, Closeable {
private final IsolationLevel isolationLevel;
private final Map<Integer, FetchSessionHandler> sessionHandlers;
private final AtomicReference<RuntimeException> cachedListOffsetsException = new AtomicReference<>();
private final Map<TopicPartition, PartitionRecords> pausedNextInLineRecordsPerTopicPartition;

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

There is an extra unnecessary space.

private final HashMap<TopicPartition, Queue<CompletedFetch>> pausedCompletedFetchesPerTopicPartition;

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

HashMap can be replaced by Map.

private final LinkedHashSet<TopicPartition> recentlyUnPausedTopicPartitions;

private PartitionRecords nextInLineRecords = null;

Expand Down Expand Up @@ -171,6 +177,10 @@ public Fetcher(LogContext logContext,
this.requestTimeoutMs = requestTimeoutMs;
this.isolationLevel = isolationLevel;
this.sessionHandlers = new HashMap<>();
this.pausedNextInLineRecordsPerTopicPartition = new HashMap<>();
this.pausedCompletedFetchesPerTopicPartition = new HashMap<>();
this.recentlyUnPausedTopicPartitions = new LinkedHashSet<>();


subscriptions.addListener(this);
}
Expand Down Expand Up @@ -198,6 +208,41 @@ public boolean hasCompletedFetches() {
return !completedFetches.isEmpty();
}

/**
* Return if there are buffered CompletedFetches for a paused TopicPartition.
*
* @param tp {@link TopicPartition}
* @return {@code true} if there are buffered {@link CompletedFetch}, {@false} otherwise
*/
public boolean hasPausedCompletedFetchesFor(TopicPartition tp) {

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Can you make this method package private by removing the keyword public and add comment that this method is used by test only?

@MayureshGharat MayureshGharat Nov 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

changed it to. : hasBufferedDataForPausedPartition(). I was thinking if there will be an usecase for apps to query if there is any buffered data for a paused partitions. I currently can't think of any. So we can make it package private for now and think of exposing it if there is a need in future.

return pausedCompletedFetchesPerTopicPartition.containsKey(tp);
}

/**
* Add the TopicPartition to the list of recently unpaused partitions.
* @param tp {@link TopicPartition}
*/
public void addRecentlyUnpausedPartition(TopicPartition tp) {
this.recentlyUnPausedTopicPartitions.add(tp);
}

/**
* Remove the TopicPartition from list of recently unpaused partitions.
* @param tp {@link TopicPartition}
*/
public void removeRecentlyUnpausedPartition(TopicPartition tp) {

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Can we simplify the patch by not having this method? We pause a partition after putting it in recentlyUnPausedTopicPartitions, we can still use fetcher.subscriptions to check whether the partition is unpaused while going over the partition in recentlyUnPausedTopicPartitions.

@MayureshGharat MayureshGharat Nov 17, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @lindong28 , can you shed more light on "We pause a partition after putting it in recentlyUnPausedTopicPartitions".
recentlyUnPausedTopicPartitions is used to give priority for returning buffered data for partitions that were paused earlier. When a partition is UnPaused, we add it to the "recentlyUnPausedTopicPartitions". When we Pause a partition, we remove it from "recentlyUnPausedPartitions".
If we have a method for adding to recentlyUnPausedPartitions, we should have a method for removing it as well, right?
The way I understand your suggested approach, if we want to use fetcher.subscriptions, we will have to iterate over all the fetchablePartitions in fetchedRecords() till we find the TopicPartition that is present in recentlyUnPausedPartitions and then check if there is data buffered for it. Iterating over fetcher.subscriptions, seems like an O(n) operation whereas using the recentlyUnpausedTopicPartitions as here, seems like O(m) operation where m < n.

@lindong28 lindong28 Nov 17, 2018 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Regarding If we have a method for adding to recentlyUnPausedPartitions, we should have a method for removing it as well, right?, there is no strict rule that requires a public API to remove entry, as long as those entries can eventually be removed. Currently fetchedRecords() will actually remove entry from recentlyUnPausedPartitions if !pausedCompletedFetchesPerTopicPartition.containsKey(tp) && !pausedNextInLineRecordsPerTopicPartition.containsKey(tp) is evaluated to true.

I agree that recentlyUnPausedPartitions is needed to improve the time complexity as you mentioned. Here the suggestion is to simplify the patch by removing the method removeRecentlyUnpausedPartition(TopicPartition tp). Is there any performance or correctness concern if we remove removeRecentlyUnpausedPartition(...)?

@MayureshGharat MayureshGharat Nov 21, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

As described above recentlyUnpausedPartitions is used to keep track of the partitions that were just unpaused before calling poll(), so that we can give preference to buffered data for previously paused partitions.
Whenever we unpause a partition, we add it to recentlyUnPausedPartitions.
So if we had completedFetch for a paused partiton P1 along with the completedFetches for unpaused partitions [P2,P3,P4], the completedFetch for paused partition P1 will be added to the pausedCompletedFetchesPerTopicPartition.
Now if we unpause this partition P1 before next poll(), it would be one of the partition eligible for returning data to the user.
But at this point we are still scraping the records from a completedFetch of some other partition and might have reached the end and the maxPollRecords limit at the same time.
So in the next poll() we are going to iterate over a new completedFetch. At this point, we would be selecting the buffered completedFetch for P1, as it is the recently unpaused partition.
But before next poll(), if we pause the partition P1 and not remove it from the recetlyUnPausedPartitions, then in fetcher.fetchedRecords(), we will see P1 as a part of recentlyUnPausedPartition and we also have the buffered data for it in pausedCompletedFetchesPerTopicPartition, so we will
iterate over this completedFetch and return data for a paused partition, which is incorrect.
Ofcourse, we can avoid this by adding a check subscriptions.isFetchable(P1) in fetcher.fetchedRecords() something like this :

if (!recentlyUnPausedTopicPartitions.isEmpty()) {
    Iterator<TopicPartition> itr = recentlyUnPausedTopicPartitions.iterator();
    TopicPartition tp = itr.next();
    if (subscriptions.isFetchable(P1)) {
    ......
    }
    ....

But I am not sure if this is less complex then just having a method to remove P1 from recentlyUnPausedPartitions. On the contrary, I feel having a method to remove P1 from recentlyUnPausedPartitions when P1 is paused is more explicit and makes the code more readable.

@lindong28 lindong28 Nov 21, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Right, the alternative solution in the code snippet above is what I am suggesting. So we agree that we can achieve the same correctness and performance without having method removeRecentlyUnpausedPartition.

All things being equal, it is always preferred to have less API (and logic) exposed from Fetcher.java. So the question here is, what is the benefit of having this extra API. I am just not convinced by the statement that this method makes code more explicit and readable. Whether all not this is more readable is an objective opinion and may be you for opinion from other developers.

@MayureshGharat MayureshGharat Nov 23, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @lindong28, thanks a lot for the comments.
I am not 100% convinced with "it is always preferred to have less API (and logic) exposed from Fetcher.java". This class is internal to the KafkaConsumer logic and not exposed by an api directly from KafkaConsumer. We already have more powerful public apis like : sendFetches(), getTopicMetadata(), fetchedRecords() exposed in this class as public methods. Also we are publicly exposing an api to add to recentlyUnPausedPartitions. So not having a method to remove from it, seems odd to me.
I agree with "Whether all not this is more readable is an objective opinion". We can wait for someone else to take a look at the discussion here and let us know there opinions. Ping : @jjkoshy @junrao @onurkaraman

this.recentlyUnPausedTopicPartitions.remove(tp);
}

/**
* Clear the data for all paused partitions.
*/
public void clearBufferedDataForPausedPartitions() {

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Can we re-use the existing method clearBufferedDataForUnassignedPartitions() and clear data as necessary in that method?

this.recentlyUnPausedTopicPartitions.clear();
this.pausedNextInLineRecordsPerTopicPartition.clear();
this.pausedCompletedFetchesPerTopicPartition.clear();
}

/**
* 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 @@ -466,6 +511,21 @@ private Map<TopicPartition, Long> beginningOrEndOffset(Collection<TopicPartition
return offsets;
}

private void readBufferedCompletedFetchesForPausedTopicPartitions(TopicPartition tp) {
Queue<CompletedFetch> pausedCompletedFetches = pausedCompletedFetchesPerTopicPartition.get(tp);
if (pausedCompletedFetches == null || pausedCompletedFetches.isEmpty()) {
pausedCompletedFetchesPerTopicPartition.remove(tp);
} else {
CompletedFetch bufferedCompletedFetch = pausedCompletedFetches.peek();

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

There is an extra space.

Also, should we just do pausedCompletedFetches.poll() here?

// remove the TopicPartition if there is no more buffered CompletedFetch left to be drained
if (pausedCompletedFetches.isEmpty()) {

@lindong28 lindong28 Oct 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This statement is always false.

pausedCompletedFetchesPerTopicPartition.remove(tp);
}
nextInLineRecords = parseCompletedFetch(bufferedCompletedFetch);
pausedCompletedFetches.poll();
}
}

/**
* Return the fetched records, empty the record buffer and update the consumed position.
*
Expand All @@ -483,24 +543,38 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> 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();
if (!recentlyUnPausedTopicPartitions.isEmpty()) {
Iterator<TopicPartition> itr = recentlyUnPausedTopicPartitions.iterator();
TopicPartition tp = itr.next();
if (pausedNextInLineRecordsPerTopicPartition.containsKey(tp)) {
nextInLineRecords = pausedNextInLineRecordsPerTopicPartition.remove(tp);
} else if (pausedCompletedFetchesPerTopicPartition.containsKey(tp)) {
readBufferedCompletedFetchesForPausedTopicPartitions(tp);
}
// remove the TopicPartition if there is no more buffered CompletedFetch or PartitionRecord left to be drained
if (!pausedCompletedFetchesPerTopicPartition.containsKey(tp) && !pausedNextInLineRecordsPerTopicPartition.containsKey(tp)) {
itr.remove();
}
throw e;
} else {
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();
}
throw e;
}
completedFetches.poll();
}
completedFetches.poll();
} else {
List<ConsumerRecord<K, V>> records = fetchRecords(nextInLineRecords, recordsRemaining);
TopicPartition partition = nextInLineRecords.partition;
Expand Down Expand Up @@ -529,13 +603,21 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
}

private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) {
boolean shouldDrainRecords = true;
if (!subscriptions.isAssigned(partitionRecords.partition)) {
// this can happen when a rebalance happened before fetched records are returned to the consumer's poll call
log.debug("Not returning fetched records for partition {} since it is no longer assigned",
partitionRecords.partition);
} else if (!subscriptions.isFetchable(partitionRecords.partition)) {
// this can happen when a partition is paused before fetched records are returned to the consumer's
// poll call or if the offset is being reset
// check if the TopicPartition is assigned but is paused. If it is paused, buffer the fetched records for
// reuse on unpause
if (subscriptions.isPaused(partitionRecords.partition)) {
this.pausedNextInLineRecordsPerTopicPartition.put(partitionRecords.partition, partitionRecords);
shouldDrainRecords = false;
nextInLineRecords = null;
}
log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable",

@zaharidichev zaharidichev Oct 31, 2018

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Maybe put a log statement, indicating that we are putting some data aside in the pausedNextInLineRecordsPerTopicPartition plus its size.

partitionRecords.partition);
} else {
Expand Down Expand Up @@ -566,7 +648,9 @@ private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecord
}
}

partitionRecords.drain();
if (shouldDrainRecords) {
partitionRecords.drain();
}
return emptyList();
}

Expand Down Expand Up @@ -855,6 +939,9 @@ private List<TopicPartition> fetchablePartitions() {
for (CompletedFetch completedFetch : completedFetches) {
exclude.add(completedFetch.partition);
}
// if there are already buffered CompletedFetches do not send another fetch for such TopicPartitions
exclude.addAll(pausedCompletedFetchesPerTopicPartition.keySet());

fetchable.removeAll(exclude);
return fetchable;
}
Expand Down Expand Up @@ -915,12 +1002,24 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
long fetchOffset = completedFetch.fetchedOffset;
PartitionRecords partitionRecords = null;
Errors error = partition.error;
boolean paused = false;

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 the topic partition is paused, buffer the completed fetch to be reuse on unpause
if (subscriptions.isAssigned(tp) && subscriptions.hasValidPosition(tp) && subscriptions.isPaused(tp)) {
Queue<CompletedFetch> pausedCompletedFetches = new ArrayDeque();
Queue<CompletedFetch> existingPausedCompletedFetches = this.pausedCompletedFetchesPerTopicPartition.putIfAbsent(tp, pausedCompletedFetches);
if (existingPausedCompletedFetches != null) {
existingPausedCompletedFetches.add(completedFetch);
} else {
pausedCompletedFetches.add(completedFetch);
}
paused = true;
}
} else if (error == Errors.NONE) {
// we are interested in this fetch only if the beginning offset matches the
// current consumed position
Expand Down Expand Up @@ -998,7 +1097,7 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
if (partitionRecords == null)
completedFetch.metricAggregator.record(tp, 0, 0);

if (error != Errors.NONE)
if (!paused && error != Errors.NONE)
// we move the partition to the end if there was an error. This way, it's more likely that partitions for
// the same topic can remain together (allowing for more efficient serialization).
subscriptions.movePartitionToEnd(tp);
Expand Down Expand Up @@ -1043,6 +1142,10 @@ private Optional<Integer> maybeLeaderEpoch(int leaderEpoch) {
@Override
public void onAssignment(Set<TopicPartition> assignment) {
sensors.updatePartitionLagAndLeadSensors(assignment);
/*
* clear the buffered data for paused partitions on new assignment

@lindong28 lindong28 Oct 31, 2018 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

We usually only add comment inside method using // rather than /* */. Also, the method name here is clear enough and the comment seems unnecessary.

*/
clearBufferedDataForPausedPartitions();
}

/**
Expand Down
16 changes: 16 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 @@ -832,6 +832,7 @@ public void testInFlightFetchOnPausedPartition() {
client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));
assertNull(fetcher.fetchedRecords().get(tp0));
assertTrue(fetcher.hasPausedCompletedFetchesFor(tp0));
}

@Test
Expand All @@ -844,6 +845,21 @@ public void testFetchOnPausedPartition() {
assertTrue(client.requests().isEmpty());
}

@Test
public void testFetchOnBufferedCompletedFetchesForPausedPartitions() {
subscriptions.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));
assertNull(fetcher.fetchedRecords().get(tp0));
assertTrue(fetcher.hasPausedCompletedFetchesFor(tp0));
assertEquals(0, fetcher.sendFetches());
}

@Test
public void testFetchNotLeaderForPartition() {
subscriptions.assignFromUser(singleton(tp0));
Expand Down