apoorvmittal10 commented on code in PR #16842:
URL: https://github.com/apache/kafka/pull/16842#discussion_r1823129269
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -540,57 +571,76 @@ void processShareFetch(ShareFetchData shareFetchData) {
return;
}
- try {
-
shareFetchData.partitionMaxBytes().keySet().forEach(topicIdPartition -> {
- SharePartitionKey sharePartitionKey = sharePartitionKey(
- shareFetchData.groupId(),
- topicIdPartition
- );
- SharePartition sharePartition =
getOrCreateSharePartition(sharePartitionKey);
+ // Initialize lazily, if required.
+ Map<TopicIdPartition, Throwable> erroneous = null;
+ Set<DelayedShareFetchKey> delayedShareFetchWatchKeys = new HashSet<>();
+ for (TopicIdPartition topicIdPartition :
shareFetchData.partitionMaxBytes().keySet()) {
+ SharePartitionKey sharePartitionKey = sharePartitionKey(
+ shareFetchData.groupId(),
+ topicIdPartition
+ );
+
+ SharePartition sharePartition;
+ try {
+ sharePartition = getOrCreateSharePartition(sharePartitionKey);
+ } catch (Exception e) {
+ // Complete the whole fetch request with an exception if there
is an error processing.
+ // The exception currently can be thrown only if there is an
error while initializing
+ // the share partition. But skip the processing for other
share partitions in the request
+ // as this situation is not expected.
+ log.error("Error processing share fetch request", e);
+ if (erroneous == null) {
+ erroneous = new HashMap<>();
Review Comment:
The exception occuring from `getOrCreateSharePartition` will be rare hence I
was of an opinion of not initializing the erroneous and do that only when
needed.
--
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]
