KAFKA-15615: Improve handling of fetching during metadata updates #15647

appchemist · 2024-04-02T15:55:00Z

Problem

As stated in the KAFKA-15615
When a fetch response receives an error about partition leadership, fencing, etc. a metadata refresh is triggered.
Until its metadata is updated, the consumer will continue to attempt fetching data for the partition, resulting in failure responses due to outdated information.

Approach

In cases where a partition requires metadata updates due to Fetch response, a temporary "Pause" approach is applied.

In this situation, the partition transitions to the AWAIT_UPDATE (ADD) state.
The AWAIT_UPDATE state is an Unfetchable state (TopicPartitionState.isFetchable() is false).
In other words, the partition is excluded from the Fetch targets in AbstractFetch.fetchablePartitions.
For KIP-951, if the fetch error response includes CurrentLeader information, the subscription remains fetchable as is.

AWAIT_UPDATE

Transitions from
- FETCHING
Transitions to
- FETCHING
- AWAIT_RESET
- AWAIT_VALIDATION
- AWAIT_UPDATE
Even if the metadata is updated without changes to Leader and Epoch information in the AWAIT_UPDATE state, the. partition transitions to the next state by SubscriptionState.maybeValidatePositionForCurrentLeader.
- FETCHING
- AWAIT_VALIDATION
requiresPosition true
hasValidPosition false

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

johnnychhsu · 2024-04-15T15:13:26Z

thanks for the PR!
just curious, what is the problem without the newly added state? does it cause any issues?

appchemist · 2024-04-15T17:29:18Z

thanks for the PR! just curious, what is the problem without the newly added state? does it cause any issues?

@johnnychhsu
When a kafka consumer encounters a FENCED_LEADER_EPOCH error or other errors requiring metadata updates, a metadata update is triggered.
However, subscription's fetchstate is FETCHING.
In this state, the consumer will continuously fetch from the old leader.(because of PreferredReadReplica is reset and metadata is not yet updated)

This patch is designed to make consumer wait without sending fetch while metadata is being updated.

appchemist · 2024-04-21T04:16:58Z

@kirktrue, PTAL, thanks in advance.

kirktrue · 2024-04-23T23:29:50Z

@appchemist—thanks for the PR, and sorry for the delay in response!

I've taken a first pass but am still working through the unit test changes.

appchemist · 2024-04-24T00:04:38Z

@kirktrue Thanks for the heads-up!
If you have a moment, please take a look

kirktrue

Thanks for the PR, @appchemist!

Just a few comments/questions.

kirktrue · 2024-04-25T20:44:36Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java

-                error == Errors.REPLICA_NOT_AVAILABLE ||
+                error == Errors.FENCED_LEADER_EPOCH) {
+            log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
+            requestMetadataUpdate(metadata, subscriptions, tp);
+        } else if (error == Errors.REPLICA_NOT_AVAILABLE ||
                error == Errors.KAFKA_STORAGE_ERROR ||
-                error == Errors.FENCED_LEADER_EPOCH ||
                error == Errors.OFFSET_NOT_AVAILABLE) {
            log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
            requestMetadataUpdate(metadata, subscriptions, tp);
+            subscriptions.awaitUpdate(tp);


With this change, if the replica is not available, we will flag the partition as awaiting a metadata update. Is this a key part of this change? Why don't we want the first if block to also await an update?

Thank you!
I missed that.

For this case, there is no need to await a metadata update.
I think simply initializing the PreferredReadReplica should be enough.
Since FetchUtils.requestMetadataUpdate() is already being called, it should also be initializing.

kirktrue · 2024-04-25T20:47:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java

+                    } else {
+                        requestMetadataUpdate(metadata, subscriptions, partition);
+                        subscriptions.awaitUpdate(partition);


With this change, we first request a metadata update, then flag our partition as awaiting the metadata update whenever we encounter a NOT_LEADER_OR_FOLLOWER or FENCED_LEADER_EPOCH. However, in the FetchCollector.handleInitializeErrors() method, we only only request the metadata update, but don't flag the partition. Is that seeming inconsistency intentional?

If the FetchStates is FETCHING as per KIP-951, the FetchCollector.handleInitializeErrors() method is called.
I thought that in this case, it should not be changed to AWAIT_UPDATE.
Additionally, if it's AWAIT_UPDATE, it will be filtered out by the following code inside the FetchCollector.initialize() method and will not go through FetchCollector.handleInitializeErrors().

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); return null; }

As another alternative, it could change the status to AWAIT_UPDATE in FetchCollector.handleInitializeErrors() only when it doesn't include leader info
Upon further thought, it seems possible to differentiate based on the following conditions.

completedFetch.partitionData.currentLeader().leaderId() != -1 && completedFetch.partitionData.currentLeader().leaderEpoch() != -1

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

appchemist · 2024-04-26T05:19:33Z

Thanks for review! @kirktrue

- add 'AWAIT_UPDATE' state in FetchStates

- Enhance tests

- AWAIT_UPDATE transitions to the next state if the Leader information in the updated metadata is the same as the existing one.

…als/SubscriptionState.java LGTM Co-authored-by: Kirk True <[email protected]>

- Just requestMetadataUpdate in REPLICA_NOT_AVAILABLE error

- Move await a metadata update logic from AbstractFetch to FetchCollector

- Add test cases about REPLICA_NOT_AVAILABLE error - Fix some comments

- Simplify if statements

kirktrue · 2024-05-02T19:20:25Z

@lianetm & @philipnee—would you be able to review this PR? Thanks!

appchemist · 2024-05-09T00:35:03Z

@lianetm & @philipnee If you have a moment, please take a look

appchemist · 2024-04-26T06:59:07Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java

-                error == Errors.KAFKA_STORAGE_ERROR ||
-                error == Errors.FENCED_LEADER_EPOCH ||
+        if (error == Errors.REPLICA_NOT_AVAILABLE) {
+            log.debug("Received replica not available error in fetch for partition {}", tp);


@kirktrue It's just a debug log, but it's different from the previous log. Is that okay?

appchemist · 2024-04-26T07:02:09Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java

+                    } else {
+                        requestMetadataUpdate(metadata, subscriptions, partition);
+                        subscriptions.awaitUpdate(partition);


appchemist · 2024-04-30T00:01:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java

-                error == Errors.REPLICA_NOT_AVAILABLE ||
+                error == Errors.FENCED_LEADER_EPOCH) {
+            log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
+            requestMetadataUpdate(metadata, subscriptions, tp);
+        } else if (error == Errors.REPLICA_NOT_AVAILABLE ||
                error == Errors.KAFKA_STORAGE_ERROR ||
-                error == Errors.FENCED_LEADER_EPOCH ||
                error == Errors.OFFSET_NOT_AVAILABLE) {
            log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
            requestMetadataUpdate(metadata, subscriptions, tp);
+            subscriptions.awaitUpdate(tp);


github-actions · 2025-03-11T03:39:30Z

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

appchemist · 2025-03-11T13:24:17Z

Hi @kirktrue
Are there any areas that need more work?
I know it's been deferred to version 4.0 for now.

appchemist changed the title ~~KAFKA-15615: Improve handling of fetching during metadata updates~~ KAFKA-15615: Improve handling of fetching during metadata updates. Apr 9, 2024

kirktrue reviewed Apr 25, 2024

View reviewed changes

appchemist changed the title ~~KAFKA-15615: Improve handling of fetching during metadata updates.~~ KAFKA-15615: Improve handling of fetching during metadata updates Apr 26, 2024

appchemist and others added 8 commits April 28, 2024 18:41

KAFKA-15615: Improve handling of fetching during metadata updates

c7a5ca8

- add 'AWAIT_UPDATE' state in FetchStates

KAFKA-15615: Improve handling of fetching during metadata updates

53825f0

- Enhance tests

KAFKA-15615: Improve handling of fetching during metadata updates

6039bae

- AWAIT_UPDATE transitions to the next state if the Leader information in the updated metadata is the same as the existing one.

Update clients/src/main/java/org/apache/kafka/clients/consumer/intern…

6b31a61

…als/SubscriptionState.java LGTM Co-authored-by: Kirk True <[email protected]>

KAFKA-15615: Improve handling of fetching during metadata updates

8b0b3f8

- Just requestMetadataUpdate in REPLICA_NOT_AVAILABLE error

KAFKA-15615: Improve handling of fetching during metadata updates

99894b0

- Move await a metadata update logic from AbstractFetch to FetchCollector

KAFKA-15615: Improve handling of fetching during metadata updates

f569d09

- Add test cases about REPLICA_NOT_AVAILABLE error - Fix some comments

KAFKA-15615: Improve handling of fetching during metadata updates

0195e86

- Simplify if statements

appchemist force-pushed the KAFKA-15615 branch from eb47ea9 to 0195e86 Compare April 28, 2024 14:34

appchemist requested a review from kirktrue April 29, 2024 15:10

appchemist commented May 10, 2024

View reviewed changes

mjsax added consumer and removed consumer labels Jul 22, 2024

philipnee added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Jul 23, 2024

kirktrue added the ctr Consumer Threading Refactor (KIP-848) label Aug 27, 2024

kirktrue added the clients label Sep 26, 2024

Merge branch 'trunk' into KAFKA-15615

eb8080a

kirktrue removed the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Dec 11, 2024

github-actions bot added the stale Stale PRs label Mar 11, 2025

github-actions bot removed the stale Stale PRs label Mar 12, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15615: Improve handling of fetching during metadata updates #15647

KAFKA-15615: Improve handling of fetching during metadata updates #15647

appchemist commented Apr 2, 2024 •

edited

Loading

johnnychhsu commented Apr 15, 2024

appchemist commented Apr 15, 2024

appchemist commented Apr 21, 2024

kirktrue commented Apr 23, 2024

appchemist commented Apr 24, 2024

kirktrue left a comment

kirktrue Apr 25, 2024

appchemist Apr 26, 2024 •

edited

Loading

appchemist Apr 30, 2024

kirktrue Apr 25, 2024

appchemist Apr 26, 2024 •

edited

Loading

appchemist Apr 26, 2024 •

edited

Loading

appchemist Apr 26, 2024

appchemist commented Apr 26, 2024

kirktrue commented May 2, 2024

appchemist commented May 9, 2024

appchemist Apr 26, 2024

appchemist Apr 26, 2024

appchemist Apr 30, 2024

github-actions bot commented Mar 11, 2025

appchemist commented Mar 11, 2025

KAFKA-15615: Improve handling of fetching during metadata updates #15647

Are you sure you want to change the base?

KAFKA-15615: Improve handling of fetching during metadata updates #15647

Conversation

appchemist commented Apr 2, 2024 • edited Loading

Problem

Approach

Committer Checklist (excluded from commit message)

johnnychhsu commented Apr 15, 2024

appchemist commented Apr 15, 2024

appchemist commented Apr 21, 2024

kirktrue commented Apr 23, 2024

appchemist commented Apr 24, 2024

kirktrue left a comment

Choose a reason for hiding this comment

kirktrue Apr 25, 2024

Choose a reason for hiding this comment

appchemist Apr 26, 2024 • edited Loading

Choose a reason for hiding this comment

appchemist Apr 30, 2024

Choose a reason for hiding this comment

kirktrue Apr 25, 2024

Choose a reason for hiding this comment

appchemist Apr 26, 2024 • edited Loading

Choose a reason for hiding this comment

appchemist Apr 26, 2024 • edited Loading

Choose a reason for hiding this comment

appchemist Apr 26, 2024

Choose a reason for hiding this comment

appchemist commented Apr 26, 2024

kirktrue commented May 2, 2024

appchemist commented May 9, 2024

appchemist Apr 26, 2024

Choose a reason for hiding this comment

appchemist Apr 26, 2024

Choose a reason for hiding this comment

appchemist Apr 30, 2024

Choose a reason for hiding this comment

github-actions bot commented Mar 11, 2025

appchemist commented Mar 11, 2025

appchemist commented Apr 2, 2024 •

edited

Loading

appchemist Apr 26, 2024 •

edited

Loading

appchemist Apr 26, 2024 •

edited

Loading

appchemist Apr 26, 2024 •

edited

Loading