KAFKA-17945, KAFKA-17944: Handling leader epoch error for offset reset APIs by apoorvmittal10 · Pull Request #17979 · apache/kafka

apoorvmittal10 · 2024-11-28T18:49:46Z

The PR sends known leader epoch while fetch offset information. If API throws exception then PR adds way to handle the exceptions and surface same to make decision for handling share partition itself.

The test for handling exception for share partition already exists in SharePartitionTest. For other changes, tests have been added in the PR.

Committer Checklist (excluded from commit message)

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

…t APIs

adixitconfluent

Thanks for the PR. Left a couple of minor comments.

adixitconfluent · 2024-12-02T06:58:26Z

-                    sharePartition.updateCacheAndOffsets(offsetForEarliestTimestamp(topicIdPartition, replicaManager));
+                    try {
+                        long startOffset = offsetForEarliestTimestamp(topicIdPartition, replicaManager, sharePartition.leaderEpoch());
+                        sharePartition.updateCacheAndOffsets(startOffset);


nit: line 90 and 91 can be merged into a single line

adixitconfluent · 2024-12-02T06:59:34Z

    /**
-     * The method is used to handle the share partition exception.
+     * The method returns a BiConsumer that handles share partition exceptions. The BiConsumer accepts
+     * a share partition key and a throwable and handles the exception.


MINOR: second "and" is extra

junrao

@apoorvmittal10 : Thanks for the PR. Just a minor comment.

junrao · 2024-12-02T19:34:37Z

-            SharePartition sharePartition = partitionCacheMap.remove(sharePartitionKey);
-            if (sharePartition != null) {
-                sharePartition.markFenced();
+    public BiConsumer<SharePartitionKey, Throwable> handleFencedSharePartitionException() {


handleFencedSharePartitionException => fencedSharePartitionHandler ?

Thanks, agree that methof name should be corrected now. Done.

adixitconfluent

Thanks for the PR. LGTM.

AndrewJSchofield

Thanks for the PR. Just one tiny nit. The removal of the SharePartitionManager reference from the DelayedShareFetch is nice.

AndrewJSchofield · 2024-12-03T15:51:30Z

-                Optional.empty(), true).timestampAndOffsetOpt();
+                Optional.of(leaderEpoch), true).timestampAndOffsetOpt();
        if (timestampAndOffset.isEmpty()) {
            throw new OffsetNotAvailableException("offset for Earliest timestamp not found for topic partition: " + topicIdPartition);


nit: Capitalisation of the message is a bit odd. Why not "Offset for earliest timestamp"? Similar command for the latest case too.

Yeah correct, something which this Pr inherited. Done.

junrao

@apoorvmittal10 : Thanks for the updated PR. LGTM. I will let @AndrewJSchofield merge the PR when he is satisfied.

AndrewJSchofield · 2024-12-03T19:57:48Z

@apoorvmittal10 : Thanks for the updated PR. LGTM. I will let @AndrewJSchofield merge the PR when he is satisfied.

Certainly. I don't have write access yet while the ASF wheels are turning, but I expect I will be able to merge in a day or two.

chia7712

@apoorvmittal10 thanks for this patch. overall LGTM.

chia7712 · 2024-12-03T19:58:22Z

+        assertTrue(resultData.isEmpty());
+        Mockito.verify(shareFetch, times(1)).addErroneous(tp0, exception);
+        Mockito.verify(exceptionHandler, times(1)).accept(any(SharePartitionKey.class), any(Throwable.class));
+        Mockito.verify(sp0, times(0)).updateCacheAndOffsets(1L);


Using any(Long.class) instead of 1L is more appropriate since we expect that updateCacheAndOffsets should not be called, regardless of the input.

Good point, done.

chia7712 · 2024-12-03T19:59:27Z

+        Throwable exception = new FencedLeaderEpochException("Fenced exception");
+        doThrow(exception).when(replicaManager).fetchOffsetForTimestamp(any(TopicPartition.class), anyLong(), any(), any(), anyBoolean());
+        when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(ShareAcquiredRecords.empty());
+        doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));


Since sp0 is a mock object, there's no need to stub updateCacheAndOffsets as it already does nothing.

Done. Removed from other places as well.

chia7712 · 2024-12-03T20:04:39Z

+
+        assertTrue(resultData.isEmpty());
+        Mockito.verify(shareFetch, times(1)).addErroneous(tp0, exception);
+        Mockito.verify(exceptionHandler, times(1)).accept(any(SharePartitionKey.class), any(Throwable.class));


Could you please replace any(SharePartitionKey.class), any(Throwable.class) by new SharePartitionKey("grp", tp0), exception

chia7712 · 2024-12-03T20:40:10Z

+                        log.error("Error while fetching offset for earliest timestamp for topicIdPartition: {}", topicIdPartition, e);
+                        shareFetch.addErroneous(topicIdPartition, e);
+                        exceptionHandler.accept(new SharePartitionKey(shareFetch.groupId(), topicIdPartition), e);
+                        // Do not fill the response for this partition and continue.


Excuse me, could you explain the difference between returning empty records and skipping partition data in the response?

So here as the topicPartition is added in errorneous so no need to fill it in response to DelayedShareFetch. When final response from SharePartitionManager will be computed then it merge the erroneous and resultant ones.

apoorvmittal10 · 2024-12-04T21:18:13Z

@AndrewJSchofield @chia7712 Please if you could re-check.

…t APIs (apache#17979) The PR sends known leader epoch while fetch offset information. If API throws exception then PR adds way to handle the exceptions and surface same to make decision for handling share partition itself. Reviewers: Abhinav Dixit <[email protected]>, Andrew Schofield <[email protected]>, Chia-Ping Tsai <[email protected]>, Jun Rao <[email protected]>

KAFKA-17945, KAFKA-17944: Handling leader epoch error for offset rese…

28ee1ae

…t APIs

apoorvmittal10 requested review from AndrewJSchofield, chia7712 and junrao November 28, 2024 18:50

github-actions Bot added core Kafka Broker KIP-932 Queues for Kafka labels Nov 28, 2024

adixitconfluent suggested changes Dec 2, 2024

View reviewed changes

junrao reviewed Dec 2, 2024

View reviewed changes

Addressing review comments

f0cfcbc

apoorvmittal10 requested review from adixitconfluent and junrao December 2, 2024 23:29

adixitconfluent approved these changes Dec 3, 2024

View reviewed changes

AndrewJSchofield requested changes Dec 3, 2024

View reviewed changes

junrao approved these changes Dec 3, 2024

View reviewed changes

chia7712 approved these changes Dec 3, 2024

View reviewed changes

Addressing review comments

dbea64c

apoorvmittal10 requested a review from AndrewJSchofield December 3, 2024 22:55

AndrewJSchofield approved these changes Dec 5, 2024

View reviewed changes

apoorvmittal10 added 3 commits December 5, 2024 11:54

Merge remote-tracking branch 'upstream/trunk' into KAFKA-17945

7bbb6cf

Merge remote-tracking branch 'upstream/trunk' into KAFKA-17945

b393a58

Merge remote-tracking branch 'upstream/trunk' into KAFKA-17945

de1b855

AndrewJSchofield merged commit ccca9f1 into apache:trunk Dec 6, 2024

peterxcli mentioned this pull request Dec 8, 2024

KAFKA-18014: Add duration based offset reset option for ShareConsumer #18096

Merged

3 tasks

Conversation

apoorvmittal10 commented Nov 28, 2024

Committer Checklist (excluded from commit message)

Uh oh!

adixitconfluent 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!

junrao 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!

adixitconfluent left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield 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!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield commented Dec 3, 2024

Uh oh!

chia7712 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!

apoorvmittal10 Dec 3, 2024 • 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!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 commented Dec 4, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

apoorvmittal10 Dec 3, 2024 •

edited

Loading