KAFKA-17108: Earliest pending upload offset #16584

abhijeetk88 · 2024-07-12T17:18:50Z

This is the first part of the implementation of KIP-1023

The purpose of this pull request is for the broker to start returning the correct offset when it receives a -6 as a timestamp in a ListOffsets API request.

Added unit tests for the new timestamp.

Committer Checklist (excluded from commit message)

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

kamalcph

Thanks for the patch! Left minor comments to address.

Should we need to add an OffsetSpec on the client side to make the earliest_pending_upload_offset queryable from command line tools similar to 6e32448?

kamalcph · 2024-08-05T09:04:43Z

core/src/main/scala/kafka/log/UnifiedLog.scala

+            val epochResult: Optional[Integer] =
+              if (leaderEpochCache.isDefined) {
+                val epochOpt = leaderEpochCache.get.epochForOffset(earliestPendingUploadOffset)
+                if (epochOpt.isPresent) {
+                  Optional.of(epochOpt.getAsInt)
+                } else {
+                  Optional.empty()
+                }
+              } else {
+                Optional.empty()
+              }
+            Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, earliestPendingUploadOffset, epochResult))


Suggested change

val epochResult: Optional[Integer] =

if (leaderEpochCache.isDefined) {

val epochOpt = leaderEpochCache.get.epochForOffset(earliestPendingUploadOffset)

if (epochOpt.isPresent) {

Optional.of(epochOpt.getAsInt)

} else {

Optional.empty()

}

} else {

Optional.empty()

}

Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, earliestPendingUploadOffset, epochResult))

val epochResult = leaderEpochCache.flatMap(_.epochForOffset(earliestPendingUploadOffset).asScala)

.asJava.asInstanceOf[Optional[Integer]]

Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, earliestPendingUploadOffset, epochResult))

Done. Thanks for the suggestion.

There are other places in the same method can benefit from this refactor as well like line 1319 and 1298

Made the requested changes. Also changed for LATEST_TIMESTAMP

kamalcph · 2024-08-05T09:14:48Z

core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala

+
+  @Test
+  def testFetchEarliestPendingUploadTimestampWithRemoteStorage(): Unit = {
+    val remoteLogManager = mock(classOf[RemoteLogManager])


we can refactor the test to extract the "prepare" phase which is duplicated across multiple tests.

abhijeetk88 · 2024-08-11T09:07:14Z

Should we need to add an OffsetSpec on the client side to make the earliest_pending_upload_offset queryable from command line tools similar to 6e32448?

@kamalcph I can make it as part of a separate change. Will that be ok?

OmniaGM · 2024-08-13T15:25:30Z

core/src/main/scala/kafka/log/UnifiedLog.scala

@@ -1334,6 +1334,27 @@ class UnifiedLog(@volatile var logStartOffset: Long,
        } else {
          Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1)))
        }
+      } else if (targetTimestamp == ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_OFFSET_TIMESTAMP) {


Can we extract this into different method and call it from here especially that the logic here has too many if stataments ? I know the rest of fetchOffsetByTimestamp has been done in-place but the method has grown and it hard to follow.

kamalcph

Overall LGTM. Left one minor comment to address. Thanks for covering the patch with detailed unit tests!

kamalcph · 2024-08-13T17:39:47Z

clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java

@@ -49,6 +49,8 @@ public class ListOffsetsRequest extends AbstractRequest {

    public static final long LATEST_TIERED_TIMESTAMP = -5L;

+    public static final long EARLIEST_PENDING_UPLOAD_OFFSET_TIMESTAMP = -6L;


consider renaming it to EARLIEST_PENDING_UPLOAD_TIMESTAMP for uniformity with other names.

abhijeetk88 · 2024-08-16T15:14:51Z

@kamalcph @OmniaGM addressed your comments. PTAL again.

kamalcph

LGTM, thanks for addressing the review comments!

Should we have to bump the ListOffsetRequest version for this change? PTAL.

cc @showuon @satishd

kamalcph · 2024-10-07T10:18:16Z

@abhijeetk88

Please rebase this PR with trunk.

github-actions · 2025-01-06T03:38:53Z

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.

abhijeetk88 · 2025-01-12T13:01:57Z

I would like to revive this PR. @kamalcph @showuon will you be able to help with reviews. I can rebase the PR with trunk.

abhijeetk88 marked this pull request as draft July 12, 2024 17:18

abhijeetk88 force-pushed the KAFKA-17108 branch from 4e50e93 to 39ab781 Compare July 23, 2024 10:21

abhijeetk88 force-pushed the KAFKA-17108 branch 6 times, most recently from b4f63a6 to d89d53b Compare August 5, 2024 06:26

abhijeetk88 marked this pull request as ready for review August 5, 2024 06:30

kamalcph added the tiered-storage Related to the Tiered Storage feature label Aug 5, 2024

kamalcph self-requested a review August 5, 2024 07:55

kamalcph reviewed Aug 5, 2024

View reviewed changes

abhijeetk88 force-pushed the KAFKA-17108 branch 3 times, most recently from e40dd55 to 5684997 Compare August 8, 2024 15:17

OmniaGM reviewed Aug 13, 2024

View reviewed changes

kamalcph reviewed Aug 13, 2024

View reviewed changes

abhijeetk88 force-pushed the KAFKA-17108 branch from 5684997 to bff7bf8 Compare August 16, 2024 14:31

KAFKA-17108: Earliest pending upload offset

5e548a3

abhijeetk88 force-pushed the KAFKA-17108 branch from bff7bf8 to 5e548a3 Compare August 16, 2024 15:11

abhijeetk88 requested review from kamalcph and OmniaGM August 16, 2024 15:15

kamalcph reviewed Aug 16, 2024

View reviewed changes

showuon self-assigned this Sep 6, 2024

github-actions bot added the stale Stale PRs label Jan 6, 2025

github-actions bot removed the stale Stale PRs label Jan 13, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17108: Earliest pending upload offset #16584

KAFKA-17108: Earliest pending upload offset #16584

abhijeetk88 commented Jul 12, 2024 •

edited

Loading

kamalcph left a comment

kamalcph Aug 5, 2024

abhijeetk88 Aug 8, 2024

OmniaGM Aug 13, 2024 •

edited

Loading

abhijeetk88 Aug 16, 2024

kamalcph Aug 5, 2024

abhijeetk88 Aug 8, 2024

abhijeetk88 commented Aug 11, 2024

OmniaGM Aug 13, 2024

abhijeetk88 Aug 16, 2024

kamalcph left a comment

kamalcph Aug 13, 2024

abhijeetk88 Aug 16, 2024

abhijeetk88 commented Aug 16, 2024

kamalcph left a comment

kamalcph commented Oct 7, 2024

github-actions bot commented Jan 6, 2025

abhijeetk88 commented Jan 12, 2025

		@@ -49,6 +49,8 @@ public class ListOffsetsRequest extends AbstractRequest {

		public static final long LATEST_TIERED_TIMESTAMP = -5L;

		public static final long EARLIEST_PENDING_UPLOAD_OFFSET_TIMESTAMP = -6L;

KAFKA-17108: Earliest pending upload offset #16584

Are you sure you want to change the base?

KAFKA-17108: Earliest pending upload offset #16584

Conversation

abhijeetk88 commented Jul 12, 2024 • edited Loading

Committer Checklist (excluded from commit message)

kamalcph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OmniaGM Aug 13, 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

abhijeetk88 commented Aug 11, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abhijeetk88 commented Aug 16, 2024

kamalcph left a comment

Choose a reason for hiding this comment

kamalcph commented Oct 7, 2024

github-actions bot commented Jan 6, 2025

abhijeetk88 commented Jan 12, 2025

abhijeetk88 commented Jul 12, 2024 •

edited

Loading

OmniaGM Aug 13, 2024 •

edited

Loading