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

Code before:
        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() {
        return completed;
    }

    /**

Review comment (bug):
After this change, the lock must be reentrant, correct? If so, we should consider replacing the `Lock` type with `ReentrantLock` to prevent potential deadlocks.

Additionally, `DelayedOperation(long delayMs, Optional<Lock> lockOpt)` seems to lose its meaning once the old group coordinator is removed. We might want to clean that up.

These points can be addressed in a follow-up.

Code after:
    }

    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() {
        return completed;
    }

    /**
     * Call-back to execute when a delayed operation gets expired and hence forced to complete.