Repository: apache/kafka
File: core/src/main/scala/kafka/log/UnifiedLog.scala (Scala)

Code before:
        } else Optional.empty[Integer]()

        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, logStartOffset, epochOpt))
      } else if (targetTimestamp == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP) {
        val curLocalLogStartOffset = localLogStartOffset()

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

        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, curLocalLogStartOffset, epochOpt))
      } else if (targetTimestamp == ListOffsetsRequest.LATEST_TIMESTAMP) {
        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, logEndOffset, latestEpochAsOptional(leaderEpochCache)))
      } else if (targetTimestamp == ListOffsetsRequest.LATEST_TIERED_TIMESTAMP) {
        if (remoteLogEnabled()) {
          val curHighestRemoteOffset = highestOffsetInRemoteStorage()

          val optEpoch: Optional[Integer] = leaderEpochCache.asJava.flatMap(cache => {
            val epoch = cache.epochForOffset(curHighestRemoteOffset)
            if (epoch.isPresent) Optional.of[Integer](epoch.getAsInt) else Optional.empty[Integer]()
          })

          Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, highestOffsetInRemoteStorage(), optEpoch))
        } else {
          Option.empty
        }
      } else if (targetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
        // Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
        // constant time access while being safe to use with concurrent collections unlike `toArray`.
        val segmentsCopy = logSegments.asScala.toBuffer
        val latestTimestampSegment = segmentsCopy.maxBy(_.maxTimestampSoFar)
        val latestTimestampAndOffset = latestTimestampSegment.readMaxTimestampAndOffsetSoFar

        Some(new TimestampAndOffset(latestTimestampAndOffset.timestamp,
          latestTimestampAndOffset.offset,
          latestEpochAsOptional(leaderEpochCache)))
      } else {
        // We need to search the first segment whose largest timestamp is >= the target timestamp if there is one.
        if (remoteLogEnabled()) {
          if (remoteLogManager.isEmpty) {
            throw new KafkaException("RemoteLogManager is empty even though the remote log storage is enabled.")
          }
          if (recordVersion.value < RecordVersion.V2.value) {
            throw new KafkaException("Tiered storage is supported only with versions supporting leader epochs, that means RecordVersion must be >= 2.")
          }

          val remoteOffset = remoteLogManager.get.findOffsetByTimestamp(topicPartition, targetTimestamp, logStartOffset, leaderEpochCache.get)
          if (remoteOffset.isPresent) {
            remoteOffset.asScala
          } else {

Review comment (performance):
If we return empty, then exception [here](https://sourcegraph.com/github.com/apache/kafka@8950ed10a7a869488d7f73821f619608caa11819/-/blob/core/src/main/scala/kafka/cluster/Partition.scala?L1594) 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))
```

Code after:
        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, logStartOffset, epochOpt))
      } else if (targetTimestamp == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP) {
        val curLocalLogStartOffset = localLogStartOffset()

        val epochResult: Optional[Integer] =
          if (leaderEpochCache.isDefined) {
            val epochOpt = leaderEpochCache.get.epochForOffset(curLocalLogStartOffset)
            if (epochOpt.isPresent) Optional.of(epochOpt.getAsInt) else Optional.empty()
          } else {
            Optional.empty()
          }

        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, curLocalLogStartOffset, epochResult))
      } else if (targetTimestamp == ListOffsetsRequest.LATEST_TIMESTAMP) {
        Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, logEndOffset, latestEpochAsOptional(leaderEpochCache)))
      } else if (targetTimestamp == ListOffsetsRequest.LATEST_TIERED_TIMESTAMP) {
        if (remoteLogEnabled()) {
          val curHighestRemoteOffset = highestOffsetInRemoteStorage()

          val epochResult: Optional[Integer] =
            if (leaderEpochCache.isDefined) {
              val epochOpt = leaderEpochCache.get.epochForOffset(curHighestRemoteOffset)
              if (epochOpt.isPresent) {
                Optional.of(epochOpt.getAsInt)
              } else if (curHighestRemoteOffset == -1) {
                Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH)
              } else {
                Optional.empty()
              }
            } else {
              Optional.empty()
            }

          Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, curHighestRemoteOffset, epochResult))
        } else {
          Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, -1L, Optional.of(-1)))
        }
      } else if (targetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
        // Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
        // constant time access while being safe to use with concurrent collections unlike `toArray`.
        val segmentsCopy = logSegments.asScala.toBuffer
        val latestTimestampSegment = segmentsCopy.maxBy(_.maxTimestampSoFar)
        val latestTimestampAndOffset = latestTimestampSegment.readMaxTimestampAndOffsetSoFar

        Some(new TimestampAndOffset(latestTimestampAndOffset.timestamp,
          latestTimestampAndOffset.offset,
          latestEpochAsOptional(leaderEpochCache)))
      } else {
        // We need to search the first segment whose largest timestamp is >= the target timestamp if there is one.
        if (remoteLogEnabled()) {
          if (remoteLogManager.isEmpty) {