OmniaGM commented on code in PR #15968:
URL: https://github.com/apache/kafka/pull/15968#discussion_r1831347260
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1410,34 +1416,34 @@ class ReplicaManager(val config: KafkaConfig,
if (traceEnabled)
trace(s"Append [$entriesPerPartition] to local log")
- entriesPerPartition.map { case (topicPartition, records) =>
-
brokerTopicStats.topicStats(topicPartition.topic).totalProduceRequestRate.mark()
+ entriesPerPartition.map { case (topicIdPartition, records) =>
+
brokerTopicStats.topicStats(topicIdPartition.topic).totalProduceRequestRate.mark()
brokerTopicStats.allTopicsStats.totalProduceRequestRate.mark()
// reject appending to internal topics if it is not allowed
- if (Topic.isInternal(topicPartition.topic) && !internalTopicsAllowed) {
- (topicPartition, LogAppendResult(
+ if (Topic.isInternal(topicIdPartition.topic) && !internalTopicsAllowed) {
+ (topicIdPartition, LogAppendResult(
LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
- Some(new InvalidTopicException(s"Cannot append to internal topic
${topicPartition.topic}")),
+ Some(new InvalidTopicException(s"Cannot append to internal topic
${topicIdPartition.topic}")),
hasCustomErrorMessage = false))
} else {
try {
- val partition = getPartitionOrException(topicPartition)
+ val partition =
getPartitionOrException(topicIdPartition.topicPartition())
Review Comment:
I added a new `partitionOrException` that accept `TopicIdPartition` and if
the `topicId` doesn't match it will return "UnknowTopicIdException`. If topic
id set to zero we will fall back to old behaviour and return the partition we
found.
--
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
