KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP by clolov · Pull Request #15213 · apache/kafka

clolov · 2024-01-17T13:09:43Z

Summary

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

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

clolov · 2024-01-17T13:11:10Z

-        val earliestLocalLogEpochEntry = leaderEpochCache.asJava.flatMap(cache => {
+        val epochOpt: Optional[Integer] = leaderEpochCache.asJava.flatMap(cache => {
          val epoch = cache.epochForOffset(curLocalLogStartOffset)
-          if (epoch.isPresent) cache.epochEntry(epoch.getAsInt) else Optional.empty[EpochEntry]()
+          if (epoch.isPresent) Optional.of[Integer](epoch.getAsInt) else Optional.empty[Integer]()
        })

-        val epochOpt = if (earliestLocalLogEpochEntry.isPresent && earliestLocalLogEpochEntry.get().startOffset <= curLocalLogStartOffset)
-          Optional.of[Integer](earliestLocalLogEpochEntry.get().epoch)
-        else Optional.empty[Integer]()
-


I didn't really see a point in this check earliestLocalLogEpochEntry.get().startOffset <= curLocalLogStartOffset. As far as I can tell the cache.epochForOffset already carries it out. Let me know in case I have misunderstood something.

This is done for backward compatibility with older message format. You can go through the comments in the previous block:

The first cached epoch usually corresponds to the log start offset, but we have to verify this since it may not be true following a message format version bump as the epoch will not be available for log entries written in the older format.

subjective:

Can we update the method similar to below for readability/clarity:

if (remoteLogEnabled()) { val curHighestRemoteOffset = highestOffsetInRemoteStorage() var result: Optional[Integer] = Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH) if (leaderEpochCache.isDefined) { val epochOpt = leaderEpochCache.get.epochForOffset(curHighestRemoteOffset) if (epochOpt.isPresent) { result = Optional.of(epochOpt.getAsInt) } } Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, highestOffsetInRemoteStorage(), result)) }

Note when the highestOffsetInRemoteStorage is -1, then there won't be corresponding leaderEpoch, we have to return the NO_PARTITION_LEADER_EPOCH constant which is -1.

Hopefully I have achieved both in the subsequent commit (reverted one of the changes and made both easier to read). Let me know if this isn't the case!

clolov · 2024-01-17T13:12:16Z

+  }
+
+  @Test
+  def testFetchLatestTieredTimestampWithRemoteStorage(): Unit = {


This test could be combined with testFetchOffsetByTimestampFromRemoteStorage as the only difference it has are lines 2167, 2193, 2203 and 2204. Let me know your thoughts!

Let's maintain two tests for clarity. Can you add comments for L2197 log.fetchOffsetByTimestamp(secondTimestamp, Some(remoteLogManager)), that it search for offset from both remote and local storage?

kamalcph · 2024-01-19T06:47:47Z

        targetTimestamp != ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP &&
-        targetTimestamp != ListOffsetsRequest.LATEST_TIMESTAMP)
+        targetTimestamp != ListOffsetsRequest.LATEST_TIMESTAMP &&
+        targetTimestamp != ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)


Should we allow the timestamp -4 and -5 when IBP is less than 0.10? Tiered Storage is supported from IBP 2.8-IV1

I have removed them both, but I don't think it would have caused problems either way.

kamalcph · 2024-01-19T07:00:32Z

-        val earliestLocalLogEpochEntry = leaderEpochCache.asJava.flatMap(cache => {
+        val epochOpt: Optional[Integer] = leaderEpochCache.asJava.flatMap(cache => {
          val epoch = cache.epochForOffset(curLocalLogStartOffset)
-          if (epoch.isPresent) cache.epochEntry(epoch.getAsInt) else Optional.empty[EpochEntry]()
+          if (epoch.isPresent) Optional.of[Integer](epoch.getAsInt) else Optional.empty[Integer]()
        })

-        val epochOpt = if (earliestLocalLogEpochEntry.isPresent && earliestLocalLogEpochEntry.get().startOffset <= curLocalLogStartOffset)
-          Optional.of[Integer](earliestLocalLogEpochEntry.get().epoch)
-        else Optional.empty[Integer]()
-


This is done for backward compatibility with older message format. You can go through the comments in the previous block:

The first cached epoch usually corresponds to the log start offset, but we have to verify this since it may not be true following a message format version bump as the epoch will not be available for log entries written in the older format.

kamalcph · 2024-01-19T08:21:33Z

+
+          Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, highestOffsetInRemoteStorage(), optEpoch))
+        } else {
+          Option.empty


If we return empty, then exception here will be thrown back to the caller. Do we want to throw an error back to the caller (or) empty response?

Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1))

Good catch! In the KIP I have specified that Kafka should return no offset in such situations. I shall aim to add an integration test from the point of view of the client in an upcoming pull request

kamalcph · 2024-01-19T08:22:45Z

+
+    log.appendAsLeader(TestUtils.singletonRecords(
+      value = TestUtils.randomBytes(10),
+      timestamp = firstTimestamp),


Why are we using the first timestamp again?

There is nothing preventing people to send batches of records with out-of-order timestamps (in fact, entries in the timeindex take this into account). I also think that for this particular case it doesn't matter as what I am trying to test here is that we don't get a response back. For the sake of focusing on what the test is supposed to test, however, I have removed this 😊

kamalcph · 2024-01-19T08:33:03Z

+  }
+
+  @Test
+  def testFetchLatestTieredTimestampWithRemoteStorage(): Unit = {


Let's maintain two tests for clarity. Can you add comments for L2197 log.fetchOffsetByTimestamp(secondTimestamp, Some(remoteLogManager)), that it search for offset from both remote and local storage?

clolov · 2024-01-23T11:58:02Z

Thanks a lot for the review @kamalcph! I have hopefully addressed everything 😊

kamalcph

LGTM, thanks for the patch!

We can add integration tests once we add support to query the latest-tiered and earliest-local offsets from the admin-client.

showuon · 2024-01-31T07:22:00Z

Will check this this or next week. Thanks.

showuon

Thanks for the PR. Left some comments.

showuon · 2024-02-01T09:33:46Z


      if (config.messageFormatVersion.isLessThan(IBP_0_10_0_IV0) &&
        targetTimestamp != ListOffsetsRequest.EARLIEST_TIMESTAMP &&
-        targetTimestamp != ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP &&


Ah, this is a bug fix, right?

Correct, Kamal called this out

showuon · 2024-02-01T09:49:35Z

-        else Optional.empty[Integer]()
+        var epochResult: Optional[Integer] = Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH)


Why do we change the return for "no leaderEpoch" case from empty to -1? I had a check, it seems won't change anything because the default value of leaderEpoch in ListOffsetsPartitionResponse is -1. Any thought on this change?

You are correct, this is a miss on my side as part of making this piece of code more readable - fixing it in the subsequent commit.

The logic should be

For EARLIEST_LOCAL_TIMESTAMP - return empty unless there is a leader epoch

For LATEST_TIERED_TIMESTAMP - if highest offset is -1 then return -1, if there is a highest offset then return empty unless there is a leader epoch

satishd · 2024-02-02T02:47:40Z

I will take a look at this PR by ~~5th~~ /7th.

clolov · 2024-02-07T14:20:51Z

Heya @showuon @kamalcph @satishd, I hope I have addressed the latest comments + rebased!

satishd

Thanks @clolov for the PR. Overall LGTM, left a couple of minor comments.

satishd

Thanks @clolov for addressing the review comments. LGTM.

showuon

LGTM!

showuon · 2024-02-19T06:09:54Z

@clolov , there's a compilation error in jdk 8/scala 2.12. Could you help fix it? Maybe rebasing to the latest trunk could solve it?
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-15213/4/pipeline

showuon · 2024-02-27T02:31:58Z

@clolov , do we have any update about the compilation error fix?

clolov · 2024-02-28T10:32:35Z

Apologies, I have had a lot of work recently, I will aim to provide an update by the end of the day today

clolov · 2024-02-28T18:11:29Z

Heya @showuon, I tried to reproduce the compilation failure of the unrelated test class locally, but have been unable to. I have rebased and kicked off a new build in case this was a temporary failure. If it occurs again I will continue looking into it.

showuon · 2024-02-29T00:19:18Z

Failed tests are unrelated.

junrao

@clolov : Thanks for the PR and sorry for the late review. Left a comment below.

junrao · 2024-04-04T17:55:45Z

     */
    public static final long EARLIEST_LOCAL_TIMESTAMP = -4L;

+    public static final long LATEST_TIERED_TIMESTAMP = -5L;


Typically, if we add a new targetTimestamp value, we will need to bump up the version of the ListOffsetsRequest. See https://github.com/apache/kafka/pull/10760/files. Otherwise, a client could be setting LATEST_TIERED_TIMESTAMP and assuming that the server supports it, but the server actually does not.

Yup, thanks a lot for bringing this up in the mailing list and here, I will open a pull request to amend this miss!

Thanks for following up, @clolov !

…e#15213) This is the first part of the implementation of KIP-1005 The purpose of this pull request is for the broker to start returning the correct offset when it receives a -5 as a timestamp in a ListOffsets API request Reviewers: Luke Chen <showuon@gmail.com>, Kamal Chandraprakash <kamal.chandraprakash@gmail.com>, Satish Duggana <satishd@apache.org>

apache#15213)" This reverts commit 55a6d30

…6400) * Revert "KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP (#15213)" * Set 3.8_IV0 as latest production version in 3.8 * Bring changes committed in KAFKA-16968 * Make ClusterTest annotation metadata default to 3.9 --------- Signed-off-by: Josep Prat <josep.prat@aiven.io> Reviewers: Justine Olshan <jolshan@confluent.io>, Jun Rao <junrao@gmail.com>

clolov requested review from kamalcph, satishd and showuon January 17, 2024 13:10

clolov commented Jan 17, 2024

View reviewed changes

kamalcph reviewed Jan 19, 2024

View reviewed changes

divijvaidya added kip Requires or implements a KIP tiered-storage Related to the Tiered Storage feature labels Jan 23, 2024

clolov force-pushed the kip-1005-kafka-16154 branch from 5d62284 to 791f786 Compare January 23, 2024 11:52

kamalcph approved these changes Jan 24, 2024

View reviewed changes

showuon reviewed Feb 1, 2024

View reviewed changes

clolov force-pushed the kip-1005-kafka-16154 branch from 791f786 to 4fa7671 Compare February 7, 2024 14:20

satishd reviewed Feb 7, 2024

View reviewed changes

Comment thread core/src/main/scala/kafka/log/UnifiedLog.scala Outdated

Comment thread core/src/main/scala/kafka/log/UnifiedLog.scala Outdated

clolov force-pushed the kip-1005-kafka-16154 branch from 4fa7671 to af4eb14 Compare February 8, 2024 13:16

satishd approved these changes Feb 18, 2024

View reviewed changes

showuon approved these changes Feb 19, 2024

View reviewed changes

clolov added 3 commits February 28, 2024 17:59

KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP

a929b4b

Address comments from first round of reviews

7270351

Address comments from second round of reviews

ac20141

clolov force-pushed the kip-1005-kafka-16154 branch from af4eb14 to ac20141 Compare February 28, 2024 18:05

showuon merged commit 55a6d30 into apache:trunk Feb 29, 2024

junrao reviewed Apr 4, 2024

View reviewed changes

clolov mentioned this pull request Apr 6, 2024

KAFKA-16480: Bump ListOffsets version, IBP version and mark last version of ListOffsets as unstable #15673

Closed

cmccabe mentioned this pull request Jun 18, 2024

KAFKA-16968: Make 3.8-IV0 a stable MetadataVersion and create 3.9-IV0 #16347

Merged

jlprat added a commit to jlprat/kafka that referenced this pull request Jun 20, 2024

Revert "KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP (

6acc59d

apache#15213)" This reverts commit 55a6d30

jlprat added a commit to jlprat/kafka that referenced this pull request Jun 30, 2024

Revert "KAFKA-16154: Broker returns offset for LATEST_TIERED_TIMESTAMP (

0093db8

apache#15213)" This reverts commit 55a6d30

kamalcph mentioned this pull request Aug 24, 2025

KAFKA-17108: Add EarliestPendingUpload offset spec in ListOffsets API #16584

Merged

clolov deleted the kip-1005-kafka-16154 branch January 27, 2026 11:42

		else Optional.empty[Integer]()
		var epochResult: Optional[Integer] = Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH)

Conversation

clolov commented Jan 17, 2024

Summary

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!

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!

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!

clolov commented Jan 23, 2024

Uh oh!

kamalcph left a comment

Choose a reason for hiding this comment

Uh oh!

showuon commented Jan 31, 2024

Uh oh!

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

satishd commented Feb 2, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

clolov commented Feb 7, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

satishd left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

satishd left a comment

Choose a reason for hiding this comment

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

showuon commented Feb 19, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

showuon commented Feb 27, 2024

Uh oh!

clolov commented Feb 28, 2024

Uh oh!

clolov commented Feb 28, 2024

Uh oh!

showuon commented Feb 29, 2024

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

satishd commented Feb 2, 2024 •

edited

Loading

clolov commented Feb 7, 2024 •

edited

Loading

showuon commented Feb 19, 2024 •

edited

Loading