adixitconfluent commented on code in PR #19437:
URL: https://github.com/apache/kafka/pull/19437#discussion_r2049540060
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -277,9 +323,15 @@ public boolean tryComplete() {
return false;
} catch (Exception e) {
log.error("Error processing delayed share fetch request", e);
- releasePartitionLocks(topicPartitionData.keySet());
- partitionsAcquired.clear();
- partitionsAlreadyFetched.clear();
+ // In case we have a remote fetch exception, we have already
released locks for partitions which have potential
+ // local log read. We do not release locks for partitions which
have a remote storage read because we need to
Review Comment:
I see, thanks @junrao for explaining it in detail. I think if we add the
same logic here in catch for the case when `remoteStorageFetchException` is
set, it should work.
```
boolean completedByMe = forceComplete();
// If invocation of forceComplete is not successful,
then that means the request is already completed
// hence release the acquired locks.
if (!completedByMe) {
releasePartitionLocks(partitionsAcquired.keySet());
}
```
--
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]
