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

Code before:
        }
      }

      if (config.messageFormatVersion.isLessThan(IBP_0_10_0_IV0) &&
        targetTimestamp != ListOffsetsRequest.EARLIEST_TIMESTAMP &&
        targetTimestamp != ListOffsetsRequest.LATEST_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()

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

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

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

          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`.

Review comment (performance):
Can we use `val` instead of `var` here?

```
        val epochResult: Optional[Integer] =
          if (leaderEpochCache.isDefined) {
            val epochOpt = leaderEpochCache.get.epochForOffset(curLocalLogStartOffset)
            if (epochOpt.isPresent) Optional.of(epochOpt.getAsInt) else Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH)
          } else Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH)

``` 

Code after:
        }
      }

      if (config.messageFormatVersion.isLessThan(IBP_0_10_0_IV0) &&
        targetTimestamp != ListOffsetsRequest.EARLIEST_TIMESTAMP &&
        targetTimestamp != ListOffsetsRequest.LATEST_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 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 {