runom opened a new pull request, #14841:
URL: https://github.com/apache/kafka/pull/14841
`KafkaApis.updateRecordConversionStats` may be called multiple times for a
request, and `request.messageConversionsTimeNanos` and
`request.temporaryMemoryBytes` are overwritten in this method.
```scala
def processingStatsCallback(processingStats: FetchResponseStats): Unit = {
processingStats.forKeyValue { (tp, info) =>
updateRecordConversionStats(request, tp, info)
}
}
```
```scala
private def updateRecordConversionStats(request: RequestChannel.Request,
tp: TopicPartition,
conversionStats:
RecordValidationStats): Unit = {
...
request.messageConversionsTimeNanos = conversionStats.conversionTimeNanos
}
request.temporaryMemoryBytes = conversionStats.temporaryMemoryBytes
}
```
So, in that case, MessageConversionsTimeMs and TemporaryMemoryBytes are not
recorded correctly.
This Pull Request has fixed the code to add up each value instead of
overwriting.
To test this behavior, this PR also added a test that sends a ProduceRequest
containing multiple partitions and calls `updateRecordConversionStats` through
`processingStatsCallback`.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
