junrao commented on code in PR #19759:
URL: https://github.com/apache/kafka/pull/19759#discussion_r2101385805
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -191,26 +191,23 @@ public void onExpiration() {
* Complete the share fetch operation by fetching records for all
partitions in the share fetch request irrespective
* of whether they have any acquired records. This is called when the
fetch operation is forced to complete either
* because records can be acquired for some partitions or due to MaxWaitMs
timeout.
+ * <p>
+ * On operation timeout, onComplete is invoked, last try occurs to acquire
partitions and read
+ * from log, if acquired. The fetch will only happen from local log and
not remote storage, on
+ * operation expiration.
Review Comment:
For better consistency, should we remove released partitions from
`partitionsAcquired` in `releasePartitionLocks()`? It seems that if we had done
that, we wouldn't get into the situation of releasing the lock more than once.
##########
server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperation.java:
##########
@@ -68,24 +67,36 @@ public DelayedOperation(long delayMs, Lock lock) {
* Return true iff the operation is completed by the caller: note that
* concurrent threads can try to complete the same operation, but only
* the first thread will succeed in completing the operation and return
- * true, others will still return false
+ * true, others will still return false.
*/
public boolean forceComplete() {
- if (completed.compareAndSet(false, true)) {
- // cancel the timeout timer
- cancel();
- onComplete();
- return true;
- } else {
+ // Do not proceed if the operation is already completed.
+ if (completed) {
Review Comment:
An implication of this change is that when forceComplete() is called inside
`tryComplete`, it always returns true. Currently, we have multiple places that
check the return value of forceComplete() inside `tryComplete`. Should we
adjust those?
--
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
