[
https://issues.apache.org/jira/browse/KAFKA-21397?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134533#comment-18134533
]
Jun Rao commented on KAFKA-21397:
---------------------------------
One way to fix this is to stop answering with an empty response when the
response could never fit. If a fetch response is bigger than the quota allows
in a whole window, send it anyway. Keep its bytes counted against the quota,
and throttle the client afterwards for as long as it takes to pay them off. The
quota still holds on average; it just can't be enforced by holding back a batch
that can't be split.
Claude suggested the following change to {{{}processResponseCallback{}}}:
{code:java}
val responseSize = fetchContext.getResponseSize(partitions, versionId)
val timeMs = time.milliseconds()
val requestThrottleTimeMs =
quotas.request.maybeRecordAndGetThrottleTimeMs(request, timeMs)
val bandwidthThrottleTimeMs =
quotas.fetch.maybeRecordAndGetThrottleTimeMs(request.session,
request.header.clientId(), responseSize, timeMs)
// A response bigger than the quota allows in a whole window would be throttled
however long the client waits,
// since its bytes are uncounted each time. Send it, keep it counted, and
throttle the client afterwards instead.
val exceedsQuotaWindow =
responseSize > quotas.fetch.maxValueInQuotaWindow(request.session,
request.header.clientId())
val maxThrottleTimeMs = math.max(bandwidthThrottleTimeMs, requestThrottleTimeMs)
val fetchResponse = if (maxThrottleTimeMs > 0 && !(exceedsQuotaWindow &&
requestThrottleTimeMs == 0)) {
// unchanged: uncount the bytes, throttle, return the empty response
...
} else {
val fetchResponse = fetchContext.updateAndGenerateResponseData(partitions,
nodeEndpoints.values.toSeq.asJava)
if (bandwidthThrottleTimeMs > 0) {
request.apiThrottleTimeMs(bandwidthThrottleTimeMs)
fetchResponse.data.setThrottleTimeMs(bandwidthThrottleTimeMs)
requestHelper.throttle(quotas.fetch, request, bandwidthThrottleTimeMs)
}
fetchResponse
}
{code}
> a single large batch bigger than the quota's allowed burst can stall
> consumers indefinitely
> -------------------------------------------------------------------------------------------
>
> Key: KAFKA-21397
> URL: https://issues.apache.org/jira/browse/KAFKA-21397
> Project: Kafka
> Issue Type: Bug
> Components: core
> Reporter: Jun Rao
> Priority: Major
>
> How the stall happens: The fetch size is capped to fit the quota. For
> consumer fetches, KafkaApis.scala:740-748 caps fetchMaxBytes at
> maxValueInQuotaWindow. That is quota × (quota.window.num − 1) ×
> quota.window.size.seconds (ClientQuotaManager.java:377-383), so about 10× the
> per-second quota with the defaults.
> The first batch ignores the cap. ReplicaManager.readFromLog reads the first
> non-empty partition with minOneMessage = true
> (ReplicaManager.scala:1896-1906).
> KafkaApis.scala:704-721 records the response size. That alone puts the client
> over its quota, so the throttle time is greater than 0. The broker then
> removes the bytes it just recorded and returns an empty throttled response.
> The consumer retries from the same offset after the throttle time. The quota
> is still configured based on the configured window sizes and doesn't change.
> The consumer gets the same empty response.
> Share fetch has the same pattern (KafkaApis.scala:3559-3561 and :4327).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
