junrao commented on code in PR #14697:
URL: https://github.com/apache/kafka/pull/14697#discussion_r1382110298
##########
core/src/test/scala/unit/kafka/log/LogValidatorTest.scala:
##########
@@ -1641,7 +1659,7 @@ class LogValidatorTest {
}
}
- def verifyRecordConversionStats(stats: RecordConversionStats,
numConvertedRecords: Int, records: MemoryRecords,
+ def verifyRecordConversionStats(stats: RecordValidationStats,
numConvertedRecords: Int, records: MemoryRecords,
Review Comment:
Should we rename `verifyRecordConversionStats` accordingly?
##########
clients/src/main/java/org/apache/kafka/common/record/RecordValidationStats.java:
##########
@@ -16,25 +16,30 @@
*/
package org.apache.kafka.common.record;
-public class RecordConversionStats {
+/**
+ * This class tracks resource usage during broker record validation for
eventual reporting in metrics.
+ * Record validation covers integrity checks on inbound data (e.g. checksum
verification), structural
+ * validation to make sure that records are well-formed, and conversion
between record formats if needed.
+ */
+public class RecordValidationStats {
Review Comment:
Could we change `toString` to use RecordValidationStats too?
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -818,9 +818,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
appendInfo.setMaxTimestamp(validateAndOffsetAssignResult.maxTimestampMs)
appendInfo.setOffsetOfMaxTimestamp(validateAndOffsetAssignResult.shallowOffsetOfMaxTimestampMs)
appendInfo.setLastOffset(offset.value - 1)
-
appendInfo.setRecordConversionStats(validateAndOffsetAssignResult.recordConversionStats)
- if (config.messageTimestampType == TimestampType.LOG_APPEND_TIME)
-
appendInfo.setLogAppendTime(validateAndOffsetAssignResult.logAppendTimeMs)
Review Comment:
Hmm, where do we set log append time in appendInfo without this? It seems
that `analyzeAndValidateRecords` initializes log append time to `NO_TIMESTAMP`
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -769,7 +769,7 @@ class ReplicaManager(val config: KafkaConfig,
entriesPerPartition: Map[TopicPartition, MemoryRecords],
responseCallback: Map[TopicPartition, PartitionResponse]
=> Unit,
delayedProduceLock: Option[Lock] = None,
- recordConversionStatsCallback: Map[TopicPartition,
RecordConversionStats] => Unit = _ => (),
+ recordConversionStatsCallback: Map[TopicPartition,
RecordValidationStats] => Unit = _ => (),
Review Comment:
Should we rename `recordConversionStatsCallback` accordingly too?
--
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
