KAFKA-7548 : KafkaConsumer should not throw away already fetched data for paused partitions. by MayureshGharat · Pull Request #5844 · apache/kafka

MayureshGharat · 2018-10-26T01:26:36Z

Today when we call KafkaConsumer.poll(), it will fetch data from Kafka asynchronously and is put in to a local buffer (completedFetches).
If now we pause some TopicPartitions and call KafkaConsumer.poll(), we might throw away any buffered data that we might have in the local buffer for these TopicPartitions. Generally, if an application is calling pause on some TopicPartitions, it is likely to resume those TopicPartitions in near future, which would require KafkaConsumer to re-issue a fetch for the same data that it had buffered earlier for these TopicPartitions. This is a wasted effort from the application's point of view.

The current patch does not throw away the already buffered data for paused partitions, but stores it in a separate buffer (pausedCompletedFetchesPerTopicPartition). If we have data in this separate buffer for a recently resumed partition, it is given higher preference as compared to other partitions.

The patch also makes sure that if we have data in this separate buffer (pausedCompletedFetchesPerTopicPartition) for a recently resumed partition, it will not send out another fetch request for that TopicPartition until that buffered data has been processed. This guarantees that we would have at the most 1 completedFetch buffered for a paused TopicPartition.

…partitions.

MayureshGharat · 2018-10-26T01:27:43Z

@cmccabe @lindong28 @jjkoshy can you please take a look.

ijuma · 2018-10-27T14:45:36Z

Cc @hachikuji

zaharidichev · 2018-10-31T19:34:43Z

+                shouldDrainRecords = false;
+                nextInLineRecords = null;
+            }
            log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable",


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

zaharidichev

LGTM

lindong28

Thanks for the patch. Left some comment.

lindong28 · 2018-10-31T21:25:15Z

    private final Map<Integer, FetchSessionHandler> sessionHandlers;
    private final AtomicReference<RuntimeException> cachedListOffsetsException = new AtomicReference<>();
+    private final  Map<TopicPartition, PartitionRecords> pausedNextInLineRecordsPerTopicPartition;
+    private final HashMap<TopicPartition, Queue<CompletedFetch>> pausedCompletedFetchesPerTopicPartition;


HashMap can be replaced by Map.

lindong28 · 2018-10-31T21:25:39Z

    private final IsolationLevel isolationLevel;
    private final Map<Integer, FetchSessionHandler> sessionHandlers;
    private final AtomicReference<RuntimeException> cachedListOffsetsException = new AtomicReference<>();
+    private final  Map<TopicPartition, PartitionRecords> pausedNextInLineRecordsPerTopicPartition;


There is an extra unnecessary space.

lindong28 · 2018-10-31T21:58:39Z

+        if (pausedCompletedFetches == null || pausedCompletedFetches.isEmpty()) {
+            pausedCompletedFetchesPerTopicPartition.remove(tp);
+        } else {
+            CompletedFetch bufferedCompletedFetch =  pausedCompletedFetches.peek();


There is an extra space.

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

lindong28 · 2018-10-31T21:59:27Z

+        } else {
+            CompletedFetch bufferedCompletedFetch =  pausedCompletedFetches.peek();
+            // remove the TopicPartition if there is no more buffered CompletedFetch left to be drained
+            if (pausedCompletedFetches.isEmpty()) {


This statement is always false.

lindong28 · 2018-10-31T22:07:47Z

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


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.

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.

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(...)?

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.

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.

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

lindong28 · 2018-10-31T22:11:50Z

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


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

lindong28 · 2018-10-31T22:19:10Z

+     * @param tp {@link TopicPartition}
+     * @return  {@code true} if there are buffered {@link CompletedFetch},  {@false} otherwise
+     */
+    public boolean hasPausedCompletedFetchesFor(TopicPartition tp) {


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

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.

lindong28 · 2018-10-31T22:20:28Z

    public void onAssignment(Set<TopicPartition> assignment) {
        sensors.updatePartitionLagAndLeadSensors(assignment);
+        /*
+         * clear the buffered data for paused partitions on new assignment


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

ijuma · 2018-11-25T22:53:45Z

Cc @hachikuji

hachikuji

@MayureshGharat I'm not sure you are still working on this, but an alternative solution you might consider would be to leave the paused data inside completedFetches. When we peek the data in fetchedRecords, you can check whether the partition is paused. If it is, it can be moved to the back of the queue. The advantage is less bookkeeping.

seglo · 2019-06-10T20:41:40Z

Since this issue has gone a little stale I thought I would take a fresh look.

@hachikuji In your last comment you made a suggestion to add any completedFetches belonging to paused partitions to the back of the queue (to be picked up later in a future poll). If we went this path then we would add the fetch to the back of the queue, poll, and peek the next record. In the case where all completedFetches belong to paused partitions we would need to confirm that they have all been checked to avoid reprocessing any fetches that were already added to the back of the queue, then return an empty set.

@MayureshGharat Do you have plans to pick up this PR again some time soon? If not, I would love the opportunity to dive into the Consumer client code to come up with a workable solution.

seglo · 2019-06-11T19:29:35Z

I spoke with @MayureshGharat in email and we agreed that I'll follow up on this work (thanks!) I'll create a new PR.

seglo · 2019-06-23T23:52:53Z

I've put up a new PR #6988

@MayureshGharat

…partitions (#6988) This is an updated implementation of #5844 by @MayureshGharat (with Mayuresh's permission). As described in the original ticket: > Today when we call KafkaConsumer.poll(), it will fetch data from Kafka asynchronously and is put in to a local buffer (completedFetches). > > If now we pause some TopicPartitions and call KafkaConsumer.poll(), we might throw away any buffered data that we might have in the local buffer for these TopicPartitions. Generally, if an application is calling pause on some TopicPartitions, it is likely to resume those TopicPartitions in near future, which would require KafkaConsumer to re-issue a fetch for the same data that it had buffered earlier for these TopicPartitions. This is a wasted effort from the application's point of view. This patch fixes the problem by retaining the paused data in the completed fetches queue, essentially moving it to the back on each call to `fetchedRecords`. Reviewers: Jason Gustafson <jason@confluent.io>

hachikuji · 2019-08-02T00:12:49Z

Going to close this since we merged #6988.

KafkaConsumer should not throw away already buffered data for paused …

f678283

…partitions.

zaharidichev reviewed Oct 31, 2018

View reviewed changes

zaharidichev approved these changes Oct 31, 2018

View reviewed changes

lindong28 reviewed Oct 31, 2018

View reviewed changes

lindong28 self-assigned this Nov 18, 2018

ennru mentioned this pull request Dec 3, 2018

Test case illustrating dropping of fetched data akka/alpakka-kafka#551

Closed

lindong28 removed their assignment Dec 7, 2018

ennru mentioned this pull request Mar 22, 2019

Adapt poll interval to demand akka/alpakka-kafka#559

Closed

hachikuji reviewed Apr 5, 2019

View reviewed changes

seglo mentioned this pull request Jun 23, 2019

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

Merged

hachikuji closed this Aug 2, 2019

Uh oh!

Conversation

MayureshGharat commented Oct 26, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MayureshGharat commented Oct 26, 2018

Uh oh!

ijuma commented Oct 27, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zaharidichev left a comment

Choose a reason for hiding this comment

Uh oh!

lindong28 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lindong28 Nov 17, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MayureshGharat Nov 21, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MayureshGharat Nov 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lindong28 Oct 31, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Nov 25, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

seglo commented Jun 10, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

seglo commented Jun 11, 2019

Uh oh!

seglo commented Jun 23, 2019

Uh oh!

hachikuji commented Aug 2, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

MayureshGharat commented Oct 26, 2018 •

edited

Loading

ijuma commented Oct 27, 2018 •

edited

Loading

lindong28 Nov 17, 2018 •

edited

Loading

MayureshGharat Nov 21, 2018 •

edited

Loading

MayureshGharat Nov 23, 2018 •

edited

Loading

lindong28 Oct 31, 2018 •

edited

Loading

seglo commented Jun 10, 2019 •

edited

Loading