KAFKA-18522: Slice records for share fetch #18804

apoorvmittal10 · 2025-02-04T19:59:57Z

The PR handles slicing of fetched records based on acquire response for share fetch. There could be additional bytes fetched from log but acquired offsets can be a subset, typically with max fetch records configuration. Rather sending additional bytes of fetched data to client we should slice the file and wire only needed batches.

Note: If the acquired offsets are within a batch then we need to send the entire batch within the file record. Hence rather checking for inidividual batches, PR finds the first and last acquired offset, and trims the file for all batches between (inclusive) these two offsets.

Copilot

Copilot reviewed 5 out of 9 changed files in this pull request and generated 1 comment.

Files not reviewed (4)

gradle/spotbugs-exclude.xml: Language not supported
core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java: Evaluated as low risk
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java: Evaluated as low risk
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java: Evaluated as low risk

Comments suppressed due to low confidence (3)

core/src/test/java/kafka/server/share/SharePartitionTest.java:5898

Consider overloading the fetchAcquiredRecords method instead of adding a boolean flag for subsetAcquired.

private List<AcquiredRecords> fetchAcquiredRecords(

core/src/test/java/kafka/server/share/SharePartitionTest.java:5914

Ensure that the memoryRecordsBuilder method from ShareFetchTestUtils is used consistently across all tests.

return memoryRecordsBuilder(numOfRecords, startOffset).build();

server/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java:45

Ensure that the subsetAcquired flag is properly tested to verify the slicing logic works correctly when this flag is true or false.

private final boolean subsetAcquired;

server/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java

Copilot

Copilot reviewed 5 out of 9 changed files in this pull request and generated 1 comment.

Files not reviewed (4)

gradle/spotbugs-exclude.xml: Language not supported
core/src/main/java/kafka/server/share/ShareFetchUtils.java: Evaluated as low risk
core/src/main/java/kafka/server/share/SharePartition.java: Evaluated as low risk
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java: Evaluated as low risk

server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTestUtils.java

clolov

Carried out an initial pass 😊, thank you, this has been an interesting read!

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

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

apoorvmittal10 · 2025-02-06T23:05:16Z

Carried out an initial pass 😊, thank you, this has been an interesting read!

Thanks for taking a look, indeed it's an intersesting change to optimize the transferred bytes.

junrao

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

junrao · 2025-02-06T20:59:48Z

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

+     *        of the fetched records. Otherwise, the original records are returned.
+     */
+    static Records maybeSliceFetchRecords(Records records, ShareAcquiredRecords shareAcquiredRecords) {
+        if (!shareAcquiredRecords.subsetAcquired() || !(records instanceof FileRecords fileRecords)) {


Hmm, with remote storage, it's possible for records to be of MemoryRecords. It would be useful to slice it too.

I see, I do not see any specific slicing API in memory records. Do you think I should add one? Or there exists some way already?

It seems there is no slicing API in memory records. So, we will need to add one.

Should I do it int this PR itself or another PR/task? Else this PR will get too long to include new API in memory records and respective individual tests, whiile integrating here. I ll prefer separately.

junrao · 2025-02-06T21:05:36Z

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

+            for (FileChannelRecordBatch batch : fileRecords.batches()) {
+                // If the batch base offset is less than the first acquired offset, then the start position
+                // should be updated to skip the batch.
+                if (batch.baseOffset() < firstAcquiredOffset) {


Hmm, not sure why we need to maintain previousBatch below. Could set just set startPosition when batch.lastOffset() is >= firstAcquiredOffset for the first time?

Yes you are correct that would have been the easiest way. But as lastOffset of batch loads headers hence I have avoided that call by maintaining previousBatch.

Thanks for explanation. Got it.

I was thinking if we could make the code a bit easier to understand. Specially, rename previousBatch to sth like mayOverlapBatch. Instead of first increasing startPosition and later decreasing it, we only increase startPosition when we are sure mayOverlapBatch indeed overlaps in the next iteration.

@junrao I re-thought about the solution. And tried to simplify it, also got complete rid of lastOffset() method call. See if it makes better sense now. Also have skipped any calculation if there exists single fetched batch, then we should not consider any slicing, as the acquired records should always be within fetched data.

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

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

junrao · 2025-02-06T21:47:39Z

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

+        assertEquals(7, recordBatches.get(0).baseOffset());
+        assertEquals(10, recordBatches.get(0).lastOffset());
+
+        // Acquire including gaps between batches, should return 2 batches.


Hmm, there are no gaps btw batches, right?

Gap is at offset 5 and 6 hence the check just validates that there occurs no issue when acquired near gap boundaries.

AndrewJSchofield

This is not an area of code that I feel qualified to review authoritatively. However, I see that the changes seem compatible with my thoughts about how to handle isolation level when we tackle that feature.

apoorvmittal10 · 2025-02-07T16:45:55Z

Thanks for the review @junrao, I have addressed and replied to the comments. I also have one doubt regrding MemoryRecords, please if you can guide.

junrao

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

junrao · 2025-02-07T19:59:10Z

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

+            for (FileChannelRecordBatch batch : fileRecords.batches()) {
+                // If the batch base offset is less than the first acquired offset, then the start position
+                // should be updated to skip the batch.
+                if (batch.baseOffset() < firstAcquiredOffset) {


Thanks for explanation. Got it.

I was thinking if we could make the code a bit easier to understand. Specially, rename previousBatch to sth like mayOverlapBatch. Instead of first increasing startPosition and later decreasing it, we only increase startPosition when we are sure mayOverlapBatch indeed overlaps in the next iteration.

server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTestUtils.java

junrao · 2025-02-08T00:31:23Z

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

+     *        of the fetched records. Otherwise, the original records are returned.
+     */
+    static Records maybeSliceFetchRecords(Records records, ShareAcquiredRecords shareAcquiredRecords) {
+        if (!shareAcquiredRecords.subsetAcquired() || !(records instanceof FileRecords fileRecords)) {


It seems there is no slicing API in memory records. So, we will need to add one.

apoorvmittal10 · 2025-02-10T22:05:52Z

Hi @junrao, can you please re-review the simplified solution.

KAFKA-18522: Slice records for share fetch

b5fdf29

apoorvmittal10 requested review from Copilot, AndrewJSchofield and junrao February 4, 2025 20:00

github-actions bot added triage PRs from the community core Kafka Broker KIP-932 Queues for Kafka build Gradle build or GitHub Actions labels Feb 4, 2025

apoorvmittal10 removed the triage PRs from the community label Feb 4, 2025

Copilot AI reviewed Feb 4, 2025

View reviewed changes

server/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java Outdated Show resolved Hide resolved

Added co-pilot suggestion

cfbfb65

apoorvmittal10 requested a review from Copilot February 4, 2025 20:15

Copilot AI reviewed Feb 4, 2025

View reviewed changes

server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTestUtils.java Outdated Show resolved Hide resolved

apoorvmittal10 added 2 commits February 5, 2025 15:27

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

9c25379

Adding changes as per upstream change

55881c5

clolov reviewed Feb 6, 2025

View reviewed changes

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

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

core/src/test/java/kafka/server/share/SharePartitionTest.java Outdated Show resolved Hide resolved

Correcting code comments

847d886

apoorvmittal10 requested a review from clolov February 6, 2025 23:02

apoorvmittal10 added the ci-approved label Feb 6, 2025

junrao reviewed Feb 6, 2025

View reviewed changes

AndrewJSchofield reviewed Feb 7, 2025

View reviewed changes

apoorvmittal10 added 2 commits February 7, 2025 16:16

Removing subset acquired changes

fcf2c47

Correcting variable name

be63e8d

apoorvmittal10 requested a review from junrao February 7, 2025 16:44

junrao reviewed Feb 8, 2025

View reviewed changes

apoorvmittal10 added 2 commits February 10, 2025 21:57

Simplifying solution

87936be

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

a8988a1

apoorvmittal10 requested a review from junrao February 10, 2025 22:05

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

1cc3e16

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18522: Slice records for share fetch #18804

KAFKA-18522: Slice records for share fetch #18804

apoorvmittal10 commented Feb 4, 2025

Copilot AI left a comment

Copilot AI left a comment

clolov left a comment

apoorvmittal10 commented Feb 6, 2025

junrao left a comment

junrao Feb 6, 2025

apoorvmittal10 Feb 7, 2025

junrao Feb 8, 2025

apoorvmittal10 Feb 10, 2025 •

edited

Loading

junrao Feb 6, 2025

apoorvmittal10 Feb 7, 2025 •

edited

Loading

junrao Feb 7, 2025

apoorvmittal10 Feb 10, 2025

junrao Feb 6, 2025

apoorvmittal10 Feb 7, 2025

AndrewJSchofield left a comment

apoorvmittal10 commented Feb 7, 2025

junrao left a comment

junrao Feb 7, 2025

junrao Feb 8, 2025

apoorvmittal10 commented Feb 10, 2025

KAFKA-18522: Slice records for share fetch #18804

Are you sure you want to change the base?

KAFKA-18522: Slice records for share fetch #18804

Conversation

apoorvmittal10 commented Feb 4, 2025

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot AI left a comment

Choose a reason for hiding this comment

clolov left a comment

Choose a reason for hiding this comment

apoorvmittal10 commented Feb 6, 2025

junrao left a 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 Feb 10, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 Feb 7, 2025 • 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

AndrewJSchofield left a comment

Choose a reason for hiding this comment

apoorvmittal10 commented Feb 7, 2025

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Feb 10, 2025

apoorvmittal10 Feb 10, 2025 •

edited

Loading

apoorvmittal10 Feb 7, 2025 •

edited

Loading