Hangleton commented on code in PR #13378:
URL: https://github.com/apache/kafka/pull/13378#discussion_r1132523221
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala:
##########
@@ -388,18 +393,27 @@ private[group] class GroupCoordinatorAdapter(
case retentionTimeMs => Some(currentTimeMs + retentionTimeMs)
}
- val partitions = new mutable.HashMap[TopicPartition, OffsetAndMetadata]()
+ val topicIds = metadata.topicNamesToIds()
+ val partitions = new mutable.HashMap[TopicIdPartition, OffsetAndMetadata]()
request.topics.forEach { topic =>
- topic.partitions.forEach { partition =>
- val tp = new TopicPartition(topic.name, partition.partitionIndex)
- partitions += tp -> createOffsetAndMetadata(
- currentTimeMs,
- partition.committedOffset,
- partition.committedLeaderEpoch,
- partition.committedMetadata,
- partition.commitTimestamp,
- expireTimeMs
- )
+ val topicId = topicIds.get(topic.name)
+ if (Uuid.ZERO_UUID.equals(topicId)) {
+ val topicPartitions = topic.partitions.asScala.map { p =>
+ new TopicPartition(topic.name, p.partitionIndex) ->
Errors.UNKNOWN_TOPIC_OR_PARTITION
+ }
+ appendToResponse(topicPartitions.toMap)
+ } else {
Review Comment:
I see, I thought we would have to use the resolved id instead of the zero
topic id. This changes the approach. Thanks for the quick check, I will fix
this.
--
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
