apoorvmittal10 commented on code in PR #17979:
URL: https://github.com/apache/kafka/pull/17979#discussion_r1866763155
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -84,7 +86,16 @@ static Map<TopicIdPartition,
ShareFetchResponseData.PartitionData> processFetchR
// response and let the client retry the fetch. This way we do
not lose out on the data that
// would be returned for other share partitions in the fetch
request.
if (fetchPartitionData.error.code() ==
Errors.OFFSET_OUT_OF_RANGE.code()) {
-
sharePartition.updateCacheAndOffsets(offsetForEarliestTimestamp(topicIdPartition,
replicaManager));
+ try {
+ long startOffset =
offsetForEarliestTimestamp(topicIdPartition, replicaManager,
sharePartition.leaderEpoch());
+ sharePartition.updateCacheAndOffsets(startOffset);
Review Comment:
Done.
##########
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.
Review Comment:
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
