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

Code before:
        targetTimestamp != ListOffsetsRequest.LATEST_TIERED_TIMESTAMP)
        throw new UnsupportedForMessageFormatException(s"Cannot search offsets based on timestamp because message format version " +
          s"for partition $topicPartition is ${config.messageFormatVersion} which is earlier than the minimum " +
          s"required version $IBP_0_10_0_IV0")

      // For the earliest and latest, we do not need to return the timestamp.
      if (targetTimestamp == ListOffsetsRequest.EARLIEST_TIMESTAMP ||
        (!remoteLogEnabled() && targetTimestamp == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)) {
        // 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.
        val earliestEpochEntry = leaderEpochCache.asJava.flatMap(_.earliestEntry())
        val epochOpt = if (earliestEpochEntry.isPresent && earliestEpochEntry.get().startOffset <= logStartOffset) {
          Optional.of[Integer](earliestEpochEntry.get().epoch)
        } 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)))

Review comment (performance):
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.

Code after:
          s"for partition $topicPartition is ${config.messageFormatVersion} which is earlier than the minimum " +
          s"required version $IBP_0_10_0_IV0")

      // For the earliest and latest, we do not need to return the timestamp.
      if (targetTimestamp == ListOffsetsRequest.EARLIEST_TIMESTAMP ||
        (!remoteLogEnabled() && targetTimestamp == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)) {
        // 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.
        val earliestEpochEntry = leaderEpochCache.asJava.flatMap(_.earliestEntry())
        val epochOpt = if (earliestEpochEntry.isPresent && earliestEpochEntry.get().startOffset <= logStartOffset) {
          Optional.of[Integer](earliestEpochEntry.get().epoch)
        } else Optional.empty[Integer]()

        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)))
        }