Repository: apache/kafka
File: core/src/main/scala/kafka/server/ReplicaManager.scala (Scala)

Code before:
          new LogReadResult(new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY),
            Optional.empty(),
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            -1L,
            OptionalLong.empty(),
            Optional.of(e)
          )
      }
    }

    var limitBytes = params.maxBytes
    val result = new mutable.ArrayBuffer[(TopicIdPartition, LogReadResult)]
    var minOneMessage = true
    readPartitionInfo.foreach { case (tp, fetchInfo) =>
      val readResult = read(tp, fetchInfo, limitBytes, minOneMessage)
      val recordBatchSize = readResult.info.records.sizeInBytes
      // Once we read from a non-empty partition, we stop ignoring request and partition level size limits
      if (recordBatchSize > 0)
        minOneMessage = false
      limitBytes = math.max(0, limitBytes - recordBatchSize)
      result += (tp -> readResult)
    }
    result
  }

  private def handleOffsetOutOfRangeError(tp: TopicIdPartition, params: FetchParams, fetchInfo: PartitionData,
                                          adjustedMaxBytes: Int, minOneMessage:
                                          Boolean, log: UnifiedLog, fetchTimeMs: Long,
                                          exception: OffsetOutOfRangeException): LogReadResult = {
    val offset = fetchInfo.fetchOffset
    // In case of offset out of range errors, handle it for tiered storage only if all the below conditions are true.
    //   1) remote log manager is enabled and it is available
    //   2) `log` instance should not be null here as that would have been caught earlier with NotLeaderOrFollowerException or ReplicaNotAvailableException.
    //   3) fetch offset is within the offset range of the remote storage layer
    if (remoteLogManager.isDefined && log != null && log.remoteLogEnabled() &&
      log.logStartOffset <= offset && offset < log.localLogStartOffset())
    {
      val highWatermark = log.highWatermark
      val leaderLogStartOffset = log.logStartOffset
      val leaderLogEndOffset = log.logEndOffset

      if (params.isFromFollower || params.isFromFuture) {
        // If it is from a follower or from a future replica, then send the offset metadata only as the data is already available in remote
        // storage and throw an error saying that this offset is moved to tiered storage.
        createLogReadResult(highWatermark, leaderLogStartOffset, leaderLogEndOffset,
          new OffsetMovedToTieredStorageException("Given offset" + offset + " is moved to tiered storage"))
      } else {
        val throttleTimeMs = remoteLogManager.get.getFetchThrottleTimeMs

Review comment (suggestion):
Hmm, currently, fetch for consumer only processes the first partition with remote storage. So, it seems that this change could return less data than truly available. For example, if `maxBytes` is 2MB, `delayedRemoteStorageFetch.get.fetchMaxBytes` is 1MB, the first 2 partitions are read from remote storage, but the 3rd partition is read locally, this change will fetch 0 bytes from the 3rd partition, but only return 1MB (for the 1st partition) in the fetch response.

Note that `readFromLog()` is also used by share fetch, which supports fetching from multiple partitions with remote storage. So, this code is correct for share fetch.

Code after:
          new LogReadResult(new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY),
            Optional.empty(),
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            UnifiedLog.UNKNOWN_OFFSET,
            -1L,
            OptionalLong.empty(),
            Optional.of(e)
          )
      }
    }

    var limitBytes = params.maxBytes
    val result = new mutable.ArrayBuffer[(TopicIdPartition, LogReadResult)]
    var minOneMessage = true
    readPartitionInfo.foreach { case (tp, fetchInfo) =>
      val readResult = read(tp, fetchInfo, limitBytes, minOneMessage)
      val recordBatchSize = readResult.info.records.sizeInBytes
      // Once we read from a non-empty partition, we stop ignoring request and partition level size limits
      if (recordBatchSize > 0)
        minOneMessage = false
      // Because we don't know how much data will be retrieved in remote fetch yet, and we don't want to block the API call
      // to query remoteLogMetadata, assume it will fetch the max bytes size of data to avoid to exceed the "fetch.max.bytes" setting.
      val estimatedRecordBatchSize = if (recordBatchSize == 0 && readResult.info.delayedRemoteStorageFetch.isPresent)
        readResult.info.delayedRemoteStorageFetch.get.fetchMaxBytes else recordBatchSize
      limitBytes = math.max(0, limitBytes - estimatedRecordBatchSize)
      result += (tp -> readResult)
    }
    result
  }

  private def handleOffsetOutOfRangeError(tp: TopicIdPartition, params: FetchParams, fetchInfo: PartitionData,
                                          adjustedMaxBytes: Int, minOneMessage:
                                          Boolean, log: UnifiedLog, fetchTimeMs: Long,
                                          exception: OffsetOutOfRangeException): LogReadResult = {
    val offset = fetchInfo.fetchOffset
    // In case of offset out of range errors, handle it for tiered storage only if all the below conditions are true.
    //   1) remote log manager is enabled and it is available
    //   2) `log` instance should not be null here as that would have been caught earlier with NotLeaderOrFollowerException or ReplicaNotAvailableException.
    //   3) fetch offset is within the offset range of the remote storage layer
    if (remoteLogManager.isDefined && log != null && log.remoteLogEnabled() &&
      log.logStartOffset <= offset && offset < log.localLogStartOffset())
    {
      val highWatermark = log.highWatermark
      val leaderLogStartOffset = log.logStartOffset
      val leaderLogEndOffset = log.logEndOffset

      if (params.isFromFollower || params.isFromFuture) {
        // If it is from a follower or from a future replica, then send the offset metadata only as the data is already available in remote
        // storage and throw an error saying that this offset is moved to tiered storage.