AndrewJSchofield commented on code in PR #17979:
URL: https://github.com/apache/kafka/pull/17979#discussion_r1867980892
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -123,11 +134,11 @@ static Map<TopicIdPartition,
ShareFetchResponseData.PartitionData> processFetchR
*
* @return The offset for the earliest timestamp.
*/
- static long offsetForEarliestTimestamp(TopicIdPartition topicIdPartition,
ReplicaManager replicaManager) {
+ static long offsetForEarliestTimestamp(TopicIdPartition topicIdPartition,
ReplicaManager replicaManager, int leaderEpoch) {
// Isolation level is only required when reading from the latest
offset hence use Option.empty() for now.
Option<FileRecords.TimestampAndOffset> timestampAndOffset =
replicaManager.fetchOffsetForTimestamp(
topicIdPartition.topicPartition(),
ListOffsetsRequest.EARLIEST_TIMESTAMP, Option.empty(),
- 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);
Review Comment:
nit: Capitalisation of the message is a bit odd. Why not "Offset for
earliest timestamp"? Similar command for the latest case too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
