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

Code before:
        // 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`.
        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.")
          }

Review comment (performance):
Can we use `val` instead of `var` here?

```
val epochResult: Optional[Integer] =
            if (leaderEpochCache.isDefined) {
              val epochOpt = leaderEpochCache.get.epochForOffset(curHighestRemoteOffset)
              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:
        // 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)))
        }
      } 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