apoorvmittal10 commented on code in PR #17979:
URL: https://github.com/apache/kafka/pull/17979#discussion_r1868468425
##########
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 {
+
sharePartition.updateCacheAndOffsets(offsetForEarliestTimestamp(topicIdPartition,
+ replicaManager, sharePartition.leaderEpoch()));
+ } catch (Exception e) {
+ log.error("Error while fetching offset for earliest
timestamp for topicIdPartition: {}", topicIdPartition, e);
+ shareFetch.addErroneous(topicIdPartition, e);
+ exceptionHandler.accept(new
SharePartitionKey(shareFetch.groupId(), topicIdPartition), e);
+ // Do not fill the response for this partition and
continue.
Review Comment:
So here as the topicPartition is added in errorneous so no need to fill it
in response to DelayedShareFetch. When final response from
SharePartitionManager will be computed then it merge the erroneous and
resultant ones.
--
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]
