apoorvmittal10 commented on code in PR #17660:
URL: https://github.com/apache/kafka/pull/17660#discussion_r1828198138
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1653,8 +1660,7 @@ class KafkaApis(val requestChannel: RequestChannel,
(txnCoordinator.partitionFor(key), TRANSACTION_STATE_TOPIC_NAME)
case CoordinatorType.SHARE =>
- // None check already done above
- (shareCoordinator.get.partitionFor(key),
SHARE_GROUP_STATE_TOPIC_NAME)
+ (shareCoordinator.foreach(coordinator =>
coordinator.partitionFor(SharePartitionKey.getInstance(key))),
SHARE_GROUP_STATE_TOPIC_NAME)
Review Comment:
For my understanding, why did we change from get to foreach?
##########
share/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java:
##########
@@ -65,6 +65,40 @@ public static SharePartitionKey getInstance(String groupId,
TopicIdPartition top
return getInstance(groupId, topicIdPartition.topicId(),
topicIdPartition.partition());
}
+
+ /**
+ * Returns a SharePartitionKey from input string of format -
groupId:topicId:partition
+ * @param key - String in format groupId:topicId:partition
+ * @return object representing SharePartitionKey
+ */
Review Comment:
Please add `exceptions` in the javadoc which can be thrown from the method.
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1644,6 +1644,13 @@ class KafkaApis(val requestChannel: RequestChannel,
if (shareCoordinator.isEmpty) {
return (Errors.INVALID_REQUEST, Node.noNode)
}
+ try {
+ val _ = SharePartitionKey.getInstance(key)
Review Comment:
If we just need to validate if the key is correct then why don't we have a
validate method and not create a instance on SharePartitionKey when it is not
required?
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -212,6 +212,7 @@ public abstract class PersisterStateManagerHandler
implements RequestCompletionH
private final BackoffManager findCoordBackoff;
protected final Logger log = LoggerFactory.getLogger(getClass());
private Consumer<ClientResponse> onCompleteCallback;
+ private SharePartitionKey partitionKey;
Review Comment:
Now when SharePartitionKey is at instance variable then we should remove
grouId, topicId, partition as instance variables.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
