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

Code before:
package org.apache.kafka.server.purgatory;

import org.apache.kafka.server.util.timer.TimerTask;

import java.util.Optional;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * An operation whose processing needs to be delayed for at most the given delayMs. For example
 * a delayed produce operation could be waiting for specified number of acks; or
 * a delayed fetch operation could be waiting for a given number of bytes to accumulate.
 * <br/>
 * The logic upon completing a delayed operation is defined in onComplete() and will be called exactly once.
 * Once an operation is completed, isCompleted() will return true. onComplete() can be triggered by either
 * forceComplete(), which forces calling onComplete() after delayMs if the operation is not yet completed,
 * or tryComplete(), which first checks if the operation can be completed or not now, and if yes calls
 * forceComplete().
 * <br/>
 * A subclass of DelayedOperation needs to provide an implementation of both onComplete() and tryComplete().
 * <br/>
 * Noted that if you add a future delayed operation that calls ReplicaManager.appendRecords() in onComplete()
 * like DelayedJoin, you must be aware that this operation's onExpiration() needs to call actionQueue.tryCompleteAction().
 */
public abstract class DelayedOperation extends TimerTask {

    private volatile boolean completed = false;

    protected final Lock lock;

    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

Review comment (suggestion):
> onComplete() can be triggered by either forceComplete(), which forces calling onComplete() after delayMs if the operation is not yet completed, or tryComplete(), which first checks if the operation can be completed or not now, and if yes calls forceComplete().

Could we adjust the comment to sth like the following?

onComplete() is called from forceComplete(), which is triggered once by either expiration if the operation is not completed after delayMs, or tryComplete() if the operation can be completed now.

Code after:
package org.apache.kafka.server.purgatory;

import org.apache.kafka.server.util.timer.TimerTask;

import java.util.Optional;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * An operation whose processing needs to be delayed for at most the given delayMs. For example
 * a delayed produce operation could be waiting for specified number of acks; or
 * a delayed fetch operation could be waiting for a given number of bytes to accumulate.
 * <br/>
 * The logic upon completing a delayed operation is defined in onComplete() and will be called exactly once.
 * Once an operation is completed, isCompleted() will return true. onComplete() is called from forceComplete(),
 * which is triggered by either expiration, if the operation is not completed after delayMs; or tryComplete(),
 * if the operation can be completed now.
 * <br/>
 * A subclass of DelayedOperation needs to provide an implementation of both onComplete() and tryComplete().
 * <br/>
 * Noted that if you add a future delayed operation that calls ReplicaManager.appendRecords() in onComplete()
 * like DelayedJoin, you must be aware that this operation's onExpiration() needs to call actionQueue.tryCompleteAction().
 */
public abstract class DelayedOperation extends TimerTask {

    private volatile boolean completed = false;

    protected final Lock lock;

    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