Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kamalcph merged 4 commits into from
Aug 27, 2025
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b9a0de9
KAFKA-17108: Earliest pending upload offset
abhijeetk88 Aug 21, 2025
5d4a7e6
Add OffsetSpec
abhijeetk88 Aug 25, 2025
35d0911
Fix failing MetadataVersionTest
abhijeetk88 Aug 25, 2025
5401eec
Address review comments
abhijeetk88 Aug 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ public class ListOffsetsRequest extends AbstractRequest {

public static final long LATEST_TIERED_TIMESTAMP = -5L;

public static final long EARLIEST_PENDING_UPLOAD_TIMESTAMP = -6L;

public static final int CONSUMER_REPLICA_ID = -1;
public static final int DEBUGGING_REPLICA_ID = -2;

Expand Down
227 changes: 226 additions & 1 deletion core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ import org.apache.kafka.server.storage.log.{FetchIsolation, UnexpectedAppendOffs
import org.apache.kafka.server.util.{KafkaScheduler, MockTime, Scheduler}
import org.apache.kafka.storage.internals.checkpoint.{LeaderEpochCheckpointFile, PartitionMetadataFile}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, Cleaner, EpochEntry, LogConfig, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogSegments, LogStartOffsetIncrementReason, LogToClean, OffsetResultHolder, OffsetsOutOfOrderException, ProducerStateManager, ProducerStateManagerConfig, RecordValidationException, UnifiedLog, VerificationGuard}
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, AsyncOffsetReader, Cleaner, EpochEntry, LogConfig, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogSegments, LogStartOffsetIncrementReason, LogToClean, OffsetResultHolder, OffsetsOutOfOrderException, ProducerStateManager, ProducerStateManagerConfig, RecordValidationException, UnifiedLog, VerificationGuard}
import org.apache.kafka.storage.internals.utils.Throttler
import org.apache.kafka.storage.log.metrics.{BrokerTopicMetrics, BrokerTopicStats}
import org.junit.jupiter.api.Assertions._
Expand Down Expand Up @@ -2416,6 +2416,193 @@ class UnifiedLogTest {
KafkaConfig.fromProps(props)
}

@Test
def testFetchEarliestPendingUploadTimestampNoRemoteStorage(): Unit = {
val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1)
val log = createLog(logDir, logConfig)

// Test initial state before any records
assertFetchOffsetBySpecialTimestamp(log, None, new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, -1, Optional.of(-1)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)

// Append records
val _ = prepareLogWithSequentialRecords(log, recordCount = 2)

// Test state after records are appended
assertFetchOffsetBySpecialTimestamp(log, None, new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_OFFSET, -1, Optional.of(-1)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)
Comment thread
kamalcph marked this conversation as resolved.
}

@Test
def testFetchEarliestPendingUploadTimestampWithRemoteStorage(): Unit = {
val logStartOffset = 0
val (remoteLogManager: RemoteLogManager, log: UnifiedLog, timestampAndEpochs: Seq[TimestampAndEpoch]) = prepare(logStartOffset)

val (firstTimestamp, firstLeaderEpoch) = (timestampAndEpochs.head.timestamp, timestampAndEpochs.head.leaderEpoch)
val (secondTimestamp, secondLeaderEpoch) = (timestampAndEpochs(1).timestamp, timestampAndEpochs(1).leaderEpoch)
val (_, thirdLeaderEpoch) = (timestampAndEpochs(2).timestamp, timestampAndEpochs(2).leaderEpoch)

doAnswer(ans => {
val timestamp = ans.getArgument(1).asInstanceOf[Long]
Optional.of(timestamp)
.filter(_ == timestampAndEpochs.head.timestamp)
.map[TimestampAndOffset](x => new TimestampAndOffset(x, 0L, Optional.of(timestampAndEpochs.head.leaderEpoch)))
}).when(remoteLogManager).findOffsetByTimestamp(ArgumentMatchers.eq(log.topicPartition),
anyLong(), anyLong(), ArgumentMatchers.eq(log.leaderEpochCache))

// Offset 0 (first timestamp) is in remote storage and deleted locally. Offset 1 (second timestamp) is in local storage.
log.updateLocalLogStartOffset(1)
log.updateHighestOffsetInRemoteStorage(0)

// In the assertions below we test that offset 0 (first timestamp) is only in remote and offset 1 (second timestamp) is in local storage.
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(firstTimestamp, 0L, Optional.of(firstLeaderEpoch))), firstTimestamp)
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(secondTimestamp, 1L, Optional.of(secondLeaderEpoch))), secondTimestamp)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 1L, Optional.of(secondLeaderEpoch)),
ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 3L, Optional.of(thirdLeaderEpoch)),
ListOffsetsRequest.LATEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 1L, Optional.of(secondLeaderEpoch)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)
}

@Test
def testFetchEarliestPendingUploadTimestampWithRemoteStorageNoLocalDeletion(): Unit = {
val logStartOffset = 0
val (remoteLogManager: RemoteLogManager, log: UnifiedLog, timestampAndEpochs: Seq[TimestampAndEpoch]) = prepare(logStartOffset)

val (firstTimestamp, firstLeaderEpoch) = (timestampAndEpochs.head.timestamp, timestampAndEpochs.head.leaderEpoch)
val (secondTimestamp, secondLeaderEpoch) = (timestampAndEpochs(1).timestamp, timestampAndEpochs(1).leaderEpoch)
val (_, thirdLeaderEpoch) = (timestampAndEpochs(2).timestamp, timestampAndEpochs(2).leaderEpoch)

// Offsets upto 1 are in remote storage
doAnswer(ans => {
val timestamp = ans.getArgument(1).asInstanceOf[Long]
Optional.of(
timestamp match {
case x if x == firstTimestamp => new TimestampAndOffset(x, 0L, Optional.of(firstLeaderEpoch))
case x if x == secondTimestamp => new TimestampAndOffset(x, 1L, Optional.of(secondLeaderEpoch))
case _ => null
}
)
}).when(remoteLogManager).findOffsetByTimestamp(ArgumentMatchers.eq(log.topicPartition),
anyLong(), anyLong(), ArgumentMatchers.eq(log.leaderEpochCache))

// Offsets 0, 1 (first and second timestamps) are in remote storage and not deleted locally.
log.updateLocalLogStartOffset(0)
log.updateHighestOffsetInRemoteStorage(1)

// In the assertions below we test that offset 0 (first timestamp) and offset 1 (second timestamp) are on both remote and local storage
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(firstTimestamp, 0L, Optional.of(firstLeaderEpoch))), firstTimestamp)
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(secondTimestamp, 1L, Optional.of(secondLeaderEpoch))), secondTimestamp)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 1L, Optional.of(secondLeaderEpoch)),
ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 3L, Optional.of(thirdLeaderEpoch)),
ListOffsetsRequest.LATEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 2L, Optional.of(thirdLeaderEpoch)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)
}

@Test
def testFetchEarliestPendingUploadTimestampNoSegmentsUploaded(): Unit = {
val logStartOffset = 0
val (remoteLogManager: RemoteLogManager, log: UnifiedLog, timestampAndEpochs: Seq[TimestampAndEpoch]) = prepare(logStartOffset)

val (firstTimestamp, firstLeaderEpoch) = (timestampAndEpochs.head.timestamp, timestampAndEpochs.head.leaderEpoch)
val (secondTimestamp, secondLeaderEpoch) = (timestampAndEpochs(1).timestamp, timestampAndEpochs(1).leaderEpoch)
val (_, thirdLeaderEpoch) = (timestampAndEpochs(2).timestamp, timestampAndEpochs(2).leaderEpoch)

// No offsets are in remote storage
doAnswer(_ => Optional.empty[TimestampAndOffset]())
.when(remoteLogManager).findOffsetByTimestamp(ArgumentMatchers.eq(log.topicPartition),
anyLong(), anyLong(), ArgumentMatchers.eq(log.leaderEpochCache))

// Offsets 0, 1, 2 (first, second and third timestamps) are in local storage only and not uploaded to remote storage.
log.updateLocalLogStartOffset(0)
log.updateHighestOffsetInRemoteStorage(-1)

// In the assertions below we test that offset 0 (first timestamp), offset 1 (second timestamp) and offset 2 (third timestamp) are only on the local storage.
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(firstTimestamp, 0L, Optional.of(firstLeaderEpoch))), firstTimestamp)
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(secondTimestamp, 1L, Optional.of(secondLeaderEpoch))), secondTimestamp)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, -1L, Optional.of(-1)),
ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 3L, Optional.of(thirdLeaderEpoch)),
ListOffsetsRequest.LATEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 0L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)
}

@Test
def testFetchEarliestPendingUploadTimestampStaleHighestOffsetInRemote(): Unit = {
val logStartOffset = 100
val (remoteLogManager: RemoteLogManager, log: UnifiedLog, timestampAndEpochs: Seq[TimestampAndEpoch]) = prepare(logStartOffset)

val (firstTimestamp, firstLeaderEpoch) = (timestampAndEpochs.head.timestamp, timestampAndEpochs.head.leaderEpoch)
val (secondTimestamp, secondLeaderEpoch) = (timestampAndEpochs(1).timestamp, timestampAndEpochs(1).leaderEpoch)
val (_, thirdLeaderEpoch) = (timestampAndEpochs(2).timestamp, timestampAndEpochs(2).leaderEpoch)

// Offsets 100, 101, 102 (first, second and third timestamps) are in local storage and not uploaded to remote storage.
// Tiered storage copy was disabled and then enabled again, because of which the remote log segments are deleted but
// the highest offset in remote storage has become stale
doAnswer(_ => Optional.empty[TimestampAndOffset]())
.when(remoteLogManager).findOffsetByTimestamp(ArgumentMatchers.eq(log.topicPartition),
anyLong(), anyLong(), ArgumentMatchers.eq(log.leaderEpochCache))

log.updateLocalLogStartOffset(100)
log.updateHighestOffsetInRemoteStorage(50)

// In the assertions below we test that offset 100 (first timestamp), offset 101 (second timestamp) and offset 102 (third timestamp) are only on the local storage.
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(firstTimestamp, 100L, Optional.of(firstLeaderEpoch))), firstTimestamp)
assertFetchOffsetByTimestamp(log, Some(remoteLogManager), Some(new TimestampAndOffset(secondTimestamp, 101L, Optional.of(secondLeaderEpoch))), secondTimestamp)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 100L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 50L, Optional.empty()),
ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 100L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 103L, Optional.of(thirdLeaderEpoch)),
ListOffsetsRequest.LATEST_TIMESTAMP)
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager),new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, 100L, Optional.of(firstLeaderEpoch)),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)
}

private def prepare(logStartOffset: Int): (RemoteLogManager, UnifiedLog, Seq[TimestampAndEpoch]) = {
val config: KafkaConfig = createKafkaConfigWithRLM
val purgatory = new DelayedOperationPurgatory[DelayedRemoteListOffsets]("RemoteListOffsets", config.brokerId)
val remoteLogManager = spy(new RemoteLogManager(config.remoteLogManagerConfig,
0,
logDir.getAbsolutePath,
"clusterId",
mockTime,
_ => Optional.empty[UnifiedLog](),
(_, _) => {},
brokerTopicStats,
new Metrics(),
Optional.empty))
remoteLogManager.setDelayedOperationPurgatory(purgatory)

val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1, remoteLogStorageEnable = true)
val log = createLog(logDir, logConfig, logStartOffset = logStartOffset, remoteStorageSystemEnable = true, remoteLogManager = Some(remoteLogManager))

// Verify earliest pending upload offset for empty log
assertFetchOffsetBySpecialTimestamp(log, Some(remoteLogManager), new TimestampAndOffset(ListOffsetsResponse.UNKNOWN_TIMESTAMP, logStartOffset, Optional.empty()),
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP)

val timestampAndEpochs = prepareLogWithSequentialRecords(log, recordCount = 3)
(remoteLogManager, log, timestampAndEpochs)
}

/**
* Test the Log truncate operations
*/
Expand Down Expand Up @@ -4786,6 +4973,44 @@ class UnifiedLogTest {

(log, segmentWithOverflow)
}

private def assertFetchOffsetByTimestamp(log: UnifiedLog, remoteLogManagerOpt: Option[RemoteLogManager], expected: Option[TimestampAndOffset], timestamp: Long): Unit = {
val remoteOffsetReader = getRemoteOffsetReader(remoteLogManagerOpt)
val offsetResultHolder = log.fetchOffsetByTimestamp(timestamp, remoteOffsetReader)
assertTrue(offsetResultHolder.futureHolderOpt.isPresent)
offsetResultHolder.futureHolderOpt.get.taskFuture.get(1, TimeUnit.SECONDS)
assertTrue(offsetResultHolder.futureHolderOpt.get.taskFuture.isDone)
assertTrue(offsetResultHolder.futureHolderOpt.get.taskFuture.get().hasTimestampAndOffset)
assertEquals(expected.get, offsetResultHolder.futureHolderOpt.get.taskFuture.get().timestampAndOffset().orElse(null))
}

private def assertFetchOffsetBySpecialTimestamp(log: UnifiedLog, remoteLogManagerOpt: Option[RemoteLogManager], expected: TimestampAndOffset, timestamp: Long): Unit = {
val remoteOffsetReader = getRemoteOffsetReader(remoteLogManagerOpt)
val offsetResultHolder = log.fetchOffsetByTimestamp(timestamp, remoteOffsetReader)
assertEquals(new OffsetResultHolder(expected), offsetResultHolder)
}

private def getRemoteOffsetReader(remoteLogManagerOpt: Option[Any]): Optional[AsyncOffsetReader] = {
remoteLogManagerOpt match {
case Some(remoteLogManager) => Optional.of(remoteLogManager.asInstanceOf[AsyncOffsetReader])
case None => Optional.empty[AsyncOffsetReader]()
}
}

private def prepareLogWithSequentialRecords(log: UnifiedLog, recordCount: Int): Seq[TimestampAndEpoch] = {
val firstTimestamp = mockTime.milliseconds()

(0 until recordCount).map { i =>
val timestampAndEpoch = TimestampAndEpoch(firstTimestamp + i, i)
log.appendAsLeader(
TestUtils.singletonRecords(value = TestUtils.randomBytes(10), timestamp = timestampAndEpoch.timestamp),
timestampAndEpoch.leaderEpoch
)
timestampAndEpoch
}
}

case class TimestampAndEpoch(timestamp: Long, leaderEpoch: Int)
}

object UnifiedLogTest {
Expand Down
27 changes: 27 additions & 0 deletions storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1667,6 +1667,8 @@ public OffsetResultHolder fetchOffsetByTimestamp(long targetTimestamp, Optional<
} else {
return new OffsetResultHolder(new FileRecords.TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1)));
}
} else if (targetTimestamp == ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP) {
return fetchEarliestPendingUploadOffset(remoteOffsetReader);
} else if (targetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
// Cache to avoid race conditions.
List<LogSegment> segments = logSegments();
Expand Down Expand Up @@ -1709,6 +1711,31 @@ public OffsetResultHolder fetchOffsetByTimestamp(long targetTimestamp, Optional<
});
}

private OffsetResultHolder fetchEarliestPendingUploadOffset(Optional<AsyncOffsetReader> remoteOffsetReader) {
if (remoteLogEnabled()) {
long curHighestRemoteOffset = highestOffsetInRemoteStorage();

if (curHighestRemoteOffset == -1L) {
if (localLogStartOffset() == logStartOffset()) {
// No segments have been uploaded yet
return fetchOffsetByTimestamp(ListOffsetsRequest.EARLIEST_TIMESTAMP, remoteOffsetReader);
} else {
// Leader currently does not know about the already uploaded segments
return new OffsetResultHolder(Optional.of(new FileRecords.TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1))));
}
} else {
long earliestPendingUploadOffset = Math.max(curHighestRemoteOffset + 1, logStartOffset());
OptionalInt epochForOffset = leaderEpochCache.epochForOffset(earliestPendingUploadOffset);
Optional<Integer> epochResult = epochForOffset.isPresent()
? Optional.of(epochForOffset.getAsInt())
: Optional.empty();
return new OffsetResultHolder(new FileRecords.TimestampAndOffset(RecordBatch.NO_TIMESTAMP, earliestPendingUploadOffset, epochResult));
}
} else {
return new OffsetResultHolder(new FileRecords.TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1)));
}
}

/**
* Checks if the log is empty.
* @return Returns True when the log is empty. Otherwise, false.
Expand Down