apoorvmittal10 commented on code in PR #17097:
URL: https://github.com/apache/kafka/pull/17097#discussion_r1749872314
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -626,6 +613,85 @@ void maybeProcessFetchQueue() {
}
}
+ private SharePartition fetchSharePartition(SharePartitionKey
sharePartitionKey) {
+ return partitionCacheMap.computeIfAbsent(sharePartitionKey,
+ k -> {
+ long start = time.hiResClockMs();
+ SharePartition partition = new SharePartition(
+ sharePartitionKey.groupId,
+ sharePartitionKey.topicIdPartition,
+ maxInFlightMessages,
+ maxDeliveryCount,
+ recordLockDurationMs,
+ timer,
+ time,
+ persister
+ );
+ this.shareGroupMetrics.partitionLoadTime(start);
+ return partition;
+ });
+ }
+
+ private void maybeCompleteInitializationWithException(
+ SharePartitionKey sharePartitionKey,
+ CompletableFuture<Map<TopicIdPartition, PartitionData>> future,
+ Throwable throwable) {
+ if (throwable instanceof LeaderNotAvailableException) {
+ log.debug("The share partition with key {} is not initialized
yet", sharePartitionKey);
+ // Do not process the fetch request for this partition as the
leader is not initialized yet.
+ // The fetch request will be retried in the next poll.
+ // TODO: Add the request to delayed fetch purgatory.
+ return;
+ }
+
+ if (throwable instanceof NotLeaderOrFollowerException || throwable
instanceof FencedStateEpochException) {
+ log.info("The share partition with key {} is fenced with message:
{}", sharePartitionKey, throwable.getMessage());
Review Comment:
Thanks, done.
--
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
