KAFKA-17022: Fix error-prone in KafkaApis#handleFetchRequest

[frankvicky]

createResponse references a variable out of scope, and so that is error-prone since it could be not initialized when executing. We should do a bit refactor to add unconvertedFetchResponse to createResponse.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

This is the root cause of the error-prone :(

[frankvicky]

I understand, mutable variables are the root of many issues. We should keep variables as immutable as possible to avoid unexpected errors.

[frankvicky]

Hi @chia7712 , I have refactor the method, PTAL 😄

[chia7712]

@frankvicky thanks for your patch

[chia7712]

we can keep it by returning the unconvertedFetchResponse from if (maxThrottleTimeMs > 0). for example:

        val unconvertedFetchResponse = if (maxThrottleTimeMs > 0) {
          request.apiThrottleTimeMs = maxThrottleTimeMs
          // Even if we need to throttle for request quota violation, we should "unrecord" the already recorded value
          // from the fetch quota because we are going to return an empty response.
          quotas.fetch.unrecordQuotaSensor(request, responseSize, timeMs)
          if (bandwidthThrottleTimeMs > requestThrottleTimeMs) {
            requestHelper.throttle(quotas.fetch, request, bandwidthThrottleTimeMs)
          } else {
            requestHelper.throttle(quotas.request, request, requestThrottleTimeMs)
          }
          // If throttling is required, return an empty response.
          fetchContext.getThrottledResponse(maxThrottleTimeMs)
        } else {
          // Get the actual response. This will update the fetch context.
          val unconvertedFetchResponse = fetchContext.updateAndGenerateResponseData(partitions)
          val responsePartitionsSize = unconvertedFetchResponse.data().responses().stream().mapToInt(_.partitions().size()).sum()
          trace(s"Sending Fetch response with partitions.size=$responsePartitionsSize, " +
            s"metadata=${unconvertedFetchResponse.sessionId}")
          unconvertedFetchResponse
        }

        // Send the response immediately.
        requestChannel.sendResponse(request, createResponse(maxThrottleTimeMs, unconvertedFetchResponse), Some(updateConversionStats))

[frankvicky]

I see, if-else is expression in Scala. Cool

[chia7712]

As there is a convertedData, it would be nice to keep using the name: unconvertedFetchResponse

[frankvicky]

I will try to refactor it 🤔

[frankvicky]

Hi @chia7712, I have make some changes based on feedbacks, PTAL 😄

[chia7712]

LGTM