OmniaGM commented on code in PR #15968:
URL: https://github.com/apache/kafka/pull/15968#discussion_r1831445186
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -623,40 +623,55 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
- val unauthorizedTopicResponses = mutable.Map[TopicPartition,
PartitionResponse]()
- val nonExistingTopicResponses = mutable.Map[TopicPartition,
PartitionResponse]()
- val invalidRequestResponses = mutable.Map[TopicPartition,
PartitionResponse]()
- val authorizedRequestInfo = mutable.Map[TopicPartition, MemoryRecords]()
+ val unauthorizedTopicResponses = mutable.Map[TopicIdPartition,
PartitionResponse]()
+ val nonExistingTopicResponses = mutable.Map[TopicIdPartition,
PartitionResponse]()
+ val invalidRequestResponses = mutable.Map[TopicIdPartition,
PartitionResponse]()
+ val authorizedRequestInfo = mutable.Map[TopicIdPartition, MemoryRecords]()
+ val topicIdToPartitionData = new mutable.ArrayBuffer[(TopicIdPartition,
ProduceRequestData.PartitionProduceData)]
+
+ produceRequest.data.topicData.forEach { topic =>
+ topic.partitionData.forEach { partition =>
+ val (topicName, topicId) = if (topic.topicId().equals(Uuid.ZERO_UUID))
{
+ (topic.name(), metadataCache.getTopicId(topic.name()))
+ } else {
+ (metadataCache.getTopicName(topic.topicId).getOrElse(topic.name),
topic.topicId())
+ }
+
+ val topicPartition = new TopicPartition(topicName, partition.index())
+ if (topicName.isEmpty)
Review Comment:
I don't think we should, if `topicId` is set to zero we should not fail we
should instead treat this as an old client request and figure out the topic id
from the broker metadata cache.
--
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
