junrao commented on code in PR #17979:
URL: https://github.com/apache/kafka/pull/17979#discussion_r1866500432
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -650,26 +651,25 @@ private void handleInitializationException(
}
/**
- * The method is used to handle the share partition exception.
+ * The method returns a BiConsumer that handles share partition
exceptions. The BiConsumer accepts
+ * a share partition key and a throwable and handles the exception.
*
- * @param sharePartitionKey The share partition key.
- * @param throwable The exception.
+ * @return A BiConsumer that handles share partition exceptions.
*/
- public void handleFencedSharePartitionException(
- SharePartitionKey sharePartitionKey,
- Throwable throwable
- ) {
- if (throwable instanceof NotLeaderOrFollowerException || throwable
instanceof FencedStateEpochException ||
- throwable instanceof GroupIdNotFoundException || throwable
instanceof UnknownTopicOrPartitionException) {
- log.info("The share partition with key {} is fenced: {}",
sharePartitionKey, throwable.getMessage());
- // The share partition is fenced hence remove the partition from
map and let the client retry.
- // But surface the error to the client so client might take some
action i.e. re-fetch
- // the metadata and retry the fetch on new leader.
- SharePartition sharePartition =
partitionCacheMap.remove(sharePartitionKey);
- if (sharePartition != null) {
- sharePartition.markFenced();
+ public BiConsumer<SharePartitionKey, Throwable>
handleFencedSharePartitionException() {
Review Comment:
handleFencedSharePartitionException => fencedSharePartitionHandler ?
--
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]
