KAFKA-17620: Simplifying share partition acquire API (kip-932) #17283

apoorvmittal10 · 2024-09-26T12:48:34Z

Simplified Share Partition Acquire API to not include CompletableFuture. The records do not need a future handling rather can be just acquired while updating internal data structure hence now need to of CompletableFuture.

Committer Checklist (excluded from commit message)

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

core/src/main/java/kafka/server/share/ShareFetchUtils.java

AndrewJSchofield · 2024-09-27T08:52:31Z

core/src/main/java/kafka/server/share/DelayedShareFetch.java

-                    releasePartitionLocks(shareFetchData.groupId(), topicPartitionData.keySet());
-                });
-
+            Map<TopicIdPartition, ShareFetchResponseData.PartitionData> result =


I think the semantics of the exception handling are different now. If processFetchResponse throws, you'll skip the release of the locks.

I might be missing something here. There is releasePartitionLocks call as well in catch block for unknowns. Though I have moved removed them from both places and moved to finally block. Is that what you meant?

Moving it to the finally block seems a bit odd because the lock isn’t set when topicPartitionData is empty.

That's true but releasePartitionLocks iterates on passed topicIdPartitions. As they would be empty then no lock will be released, as expected.

But I think your comment is less related to behaviour but more related to that do we need to call the release if topicIdPartitions is empty, correct? Though I can add a check in releasePartitionLocks for emptiness of topicIdPartitions but that seemed to me redundant as anyways the forEach will not be iterated (also topicIdPartitions can never be null). Please let me know what you think.

chia7712 · 2024-09-27T10:51:45Z

Could you please clarify why you decided not to use CompletableFuture (#16274 (comment))?

apoorvmittal10 · 2024-09-27T12:23:19Z

Could you please clarify why you decided not to use CompletableFuture (#16274 (comment))?

Hi @chia7712, though my inclination was towards keeping the code with CompletableFuture as in future if we think about persisting state when records are acquired then it will give us flexibility. But given the code with CompletableFuture adds a lot more complexity in handling scenarios, as again pointed out during introducing purgatory, #17177 (comment). Hence it's better to remove it for now. Once we have concrete use case then we should think of adding it back.

apoorvmittal10 · 2024-09-27T12:23:51Z

Thanks @AndrewJSchofield and @chia7712 for review, I have addressed the feedback.

junrao

@apoorvmittal10 : Thanks for the PR. Left a few comments.

junrao · 2024-09-30T22:36:21Z

core/src/main/java/kafka/server/share/ShareFetchUtils.java

+                if (acquiredRecords.isEmpty()) {
+                    partitionData
+                        .setRecords(null)
+                        .setErrorCode(Errors.NONE.code())


Do we need to explicitly set errorCode since it defaults to 0? If so, it would be useful to explicitly set other fields like errorMessage too. Ditto in the else clause.

junrao · 2024-09-30T22:37:47Z

core/src/main/java/kafka/server/share/ShareFetchUtils.java

@@ -46,59 +47,59 @@ public class ShareFetchUtils {

    // Process the replica manager fetch response to update share partitions and futures. We acquire the fetched data
    // from share partitions.
-    static CompletableFuture<Map<TopicIdPartition, ShareFetchResponseData.PartitionData>> processFetchResponse(
+    static Map<TopicIdPartition, ShareFetchResponseData.PartitionData> processFetchResponse(


Should we remove futures in the comment above?

junrao · 2024-09-30T22:48:17Z

core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java

+            new SimpleRecord(null, "value".getBytes()));
+
+        Map<TopicIdPartition, FetchPartitionData> responseData = Collections.singletonMap(
+            tp0, new FetchPartitionData(Errors.OFFSET_OUT_OF_RANGE, 0L, 0L,


Hmm, the purpose of the test is to test the part when no records can be acquired. However, by passing in an error response, we will skip the acquire call.

My bad, I have added other scenrio with empty as well.

apoorvmittal10 · 2024-10-09T16:28:20Z

@junrao @chia7712 Sorry for the delay, I was away. Can you please review once again, I have addressed the comments.

junrao

@apoorvmittal10 : Thanks for the updated PR. A couple of more comments.

core/src/main/java/kafka/server/share/ShareFetchUtils.java

junrao

@apoorvmittal10 : Thanks for the updated. The changes LGTM. Could you rebase?

apoorvmittal10 · 2024-10-09T21:20:59Z

@apoorvmittal10 : Thanks for the updated. The changes LGTM. Could you rebase?

Thanks @junrao for the review. I have rebased it.

junrao

@apoorvmittal10 : Thanks for rebasing. Just one minor comment.

core/src/main/java/kafka/server/share/DelayedShareFetch.java

chia7712

@apoorvmittal10 thanks for this patch. overall LGTM

chia7712 · 2024-10-10T08:50:39Z

core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java

+        assertTrue(resultData.get(tp0).acquiredRecords().isEmpty());
+        assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
+
+        Mockito.verify(sp0, times(1)).updateCacheAndOffsets(any(Long.class));


nit: fetchOffsetForTimestamp is mock, so we can verify the actual value 1 instead of any(Long.class)

junrao

@apoorvmittal10 : Thanks for the updated PR. LGTM

KAFKA-17620: Simplifying share fetch acquire API (kip-932)

6b32132

github-actions bot added the core Kafka Broker label Sep 26, 2024

apoorvmittal10 added the KIP-932 Queues for Kafka label Sep 26, 2024

apoorvmittal10 requested a review from mumrah September 26, 2024 12:49

Restructured code further

a012563

AndrewJSchofield suggested changes Sep 27, 2024

View reviewed changes

apoorvmittal10 added 2 commits September 27, 2024 12:43

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

487f3f2

Addressing review comments

8d571bb

apoorvmittal10 requested a review from AndrewJSchofield September 27, 2024 12:23

apoorvmittal10 requested a review from chia7712 September 30, 2024 14:39

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

9fa0fa7

apoorvmittal10 mentioned this pull request Sep 30, 2024

KAFKA-17774: Adding capability to handle max fetch records in Share Fetch (KIP-932) #17322

Merged

3 tasks

junrao reviewed Sep 30, 2024

View reviewed changes

apoorvmittal10 added 3 commits October 9, 2024 15:59

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

8d4f9b0

Resolving merge conflicts

da26892

Addressing review comments

7f0b006

apoorvmittal10 requested a review from junrao October 9, 2024 16:28

junrao reviewed Oct 9, 2024

View reviewed changes

core/src/main/java/kafka/server/share/ShareFetchUtils.java Show resolved Hide resolved

core/src/main/java/kafka/server/share/ShareFetchUtils.java Outdated Show resolved Hide resolved

Addressing review comments

348260d

apoorvmittal10 requested a review from junrao October 9, 2024 17:54

junrao reviewed Oct 9, 2024

View reviewed changes

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

eb53ae4

apoorvmittal10 requested a review from junrao October 9, 2024 21:20

junrao reviewed Oct 9, 2024

View reviewed changes

core/src/main/java/kafka/server/share/DelayedShareFetch.java Show resolved Hide resolved

Removing the comment

8b9d812

apoorvmittal10 requested a review from junrao October 10, 2024 08:22

AndrewJSchofield approved these changes Oct 10, 2024

View reviewed changes

chia7712 approved these changes Oct 10, 2024

View reviewed changes

Addressing review comments

38454b7

junrao approved these changes Oct 10, 2024

View reviewed changes

junrao merged commit b98aee3 into apache:trunk Oct 10, 2024
6 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17620: Simplifying share partition acquire API (kip-932) #17283

KAFKA-17620: Simplifying share partition acquire API (kip-932) #17283

apoorvmittal10 commented Sep 26, 2024

AndrewJSchofield Sep 27, 2024

apoorvmittal10 Sep 27, 2024

chia7712 Sep 28, 2024

apoorvmittal10 Sep 30, 2024

chia7712 commented Sep 27, 2024

apoorvmittal10 commented Sep 27, 2024

apoorvmittal10 commented Sep 27, 2024

junrao left a comment

junrao Sep 30, 2024 •

edited

Loading

apoorvmittal10 Oct 9, 2024

junrao Sep 30, 2024

apoorvmittal10 Oct 9, 2024

junrao Sep 30, 2024

apoorvmittal10 Oct 9, 2024

apoorvmittal10 commented Oct 9, 2024

junrao left a comment

junrao left a comment

apoorvmittal10 commented Oct 9, 2024

junrao left a comment

chia7712 left a comment

chia7712 Oct 10, 2024

apoorvmittal10 Oct 10, 2024

junrao left a comment

KAFKA-17620: Simplifying share partition acquire API (kip-932) #17283

KAFKA-17620: Simplifying share partition acquire API (kip-932) #17283

Conversation

apoorvmittal10 commented Sep 26, 2024

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Sep 27, 2024

apoorvmittal10 commented Sep 27, 2024

apoorvmittal10 commented Sep 27, 2024

junrao left a comment

Choose a reason for hiding this comment

junrao Sep 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 9, 2024

junrao left a comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 9, 2024

junrao left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

junrao Sep 30, 2024 •

edited

Loading