[KAFKA-15894] MessageConversionsTimeMs and TemporaryMemoryBytes may not be recorded correctly.

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: core, metrics
• Assignee: Minoru Tomioka
• Reporter: Minoru Tomioka
• Created: 25/Nov/23 16:54
• Updated: 25/Feb/24 03:33

Description

KafkaApis.updateRecordConversionStats may be called multiple times for a request, and request.messageConversionsTimeNanos and request.temporaryMemoryBytes are overwritten in this method.

https://github.com/apache/kafka/blob/95f41d59b389b6f25000b7bc4ddb948cfdb90448/core/src/main/scala/kafka/server/KafkaApis.scala#L3777-L3779

private def updateRecordConversionStats(request: RequestChannel.Request,
                                          tp: TopicPartition,
                                          conversionStats: RecordValidationStats): Unit = {
...
    request.messageConversionsTimeNanos = conversionStats.conversionTimeNanos
  }
  request.temporaryMemoryBytes = conversionStats.temporaryMemoryBytes
}

 

https://github.com/apache/kafka/blob/95f41d59b389b6f25000b7bc4ddb948cfdb90448/core/src/main/scala/kafka/server/KafkaApis.scala#L706-L708

def processingStatsCallback(processingStats: FetchResponseStats): Unit = {
  processingStats.forKeyValue { (tp, info) =>
    updateRecordConversionStats(request, tp, info)
  }
}

 

So, in this case, MessageConversionsTimeMs and TemporaryMemoryBytes are not recorded correctly.

Issue Links

• links to: GitHub Pull Request #14841