Repository: apache/kafka
File: server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedOperation.java (Java)

Code before:
    public DelayedOperation(long delayMs, Optional<Lock> lockOpt) {
        this(delayMs, lockOpt.orElse(new ReentrantLock()));
    }

    public DelayedOperation(long delayMs) {
        this(delayMs, new ReentrantLock());
    }

    public DelayedOperation(long delayMs, Lock lock) {
        super(delayMs);
        this.lock = lock;
    }

    /*
     * Force completing the delayed operation, if not already completed.
     * This function can be triggered when
     *
     * 1. The operation has been verified to be completable inside tryComplete()
     * 2. The operation has expired and hence needs to be completed right now
     *
     * 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.
     */
    public boolean forceComplete() {
        // Do not proceed if the operation is already completed.
        if (completed) {
            return false;
        }
        // Attain lock prior completing the request.
        lock.lock();
        try {
            // Re-check, if the operation is already completed by some other thread.
            if (!completed) {
                completed = true;
                // cancel the timeout timer
                cancel();
                onComplete();
                return true;
            } else {
                return false;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * Check if the delayed operation is already completed
     */

Review comment (suggestion):
Only the expiration logic checks the return value of forceComplete(). We could do the following in the expiration logic and change `forceComplete()` to return void and avoid using the lock there (the caller will get the lock instead). 

```
        lock.lock();
        try {
            if (!isCompleted()) {
              forceComplete();
              onExpiration();
             }
        } finally {
            lock.unlock();
        }

```
 

Code after:
        this(delayMs, lockOpt.orElse(new ReentrantLock()));
    }

    public DelayedOperation(long delayMs) {
        this(delayMs, new ReentrantLock());
    }

    public DelayedOperation(long delayMs, Lock lock) {
        super(delayMs);
        this.lock = lock;
    }

    /*
     * Force completing the delayed operation, if not already completed.
     * This function can be triggered when
     *
     * 1. The operation has been verified to be completable inside tryComplete()
     * 2. The operation has expired and hence needs to be completed right now
     *
     * 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.
     */
    public boolean forceComplete() {
        // Do not proceed if the operation is already completed.
        if (completed) {
            return false;
        }
        // Attain lock prior completing the request.
        lock.lock();
        try {
            // Re-check, if the operation is already completed by some other thread.
            if (!completed) {
                completed = true;
                // cancel the timeout timer
                cancel();
                onComplete();
                return true;
            } else {
                return false;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * Check if the delayed operation is already completed
     */
    public boolean isCompleted() {