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

Code before:
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
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 once 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;
    }

Review comment (suggestion):
Actually, it's possible for forceComplete() to be called once from tryComplete() and another from expiration. So, we need to remove `once`. 

Code after:
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
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;
    }