Source code
package org.apache.kafka.server.purgatory;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.utils.internals.LogContext;
import org.slf4j.Logger;
import java.util.List;
import java.util.concurrent.CompletableFuture;
/**
* A delayed operation using CompletionFutures that can be created by KafkaApis and watched
* in a DelayedFuturePurgatory purgatory. This is used for ACL updates using async Authorizers.
*/
public class DelayedFuture<T> extends DelayedOperation {
private final Logger log = new LogContext().logger(DelayedFuture.class.getName());
private final List<CompletableFuture<T>> futures;
private final Runnable responseCallback;
private final long timeoutMs;
public DelayedFuture(long timeoutMs, List<CompletableFuture<T>> futures, Runnable responseCallback) {
super(timeoutMs);
this.timeoutMs = timeoutMs;
this.futures = futures;
this.responseCallback = responseCallback;
}
/**
* The operation can be completed if all the futures have completed successfully
* or failed with exceptions.
*/
@Override
public boolean tryComplete() {
log.trace("Trying to complete operation for {} futures", futures.size());
long pending = futures.stream().filter(future -> !future.isDone()).count();
if (pending == 0x0) {
log.trace("All futures have been completed or have errors, completing the delayed operation");
return forceComplete();
} else {
log.trace("{} future still pending, not completing the delayed operation", pending);
return false;
}
}
/**
* Timeout any pending futures and invoke responseCallback. This is invoked when all
* futures have completed or the operation has timed out.
*/
@Override
public void onComplete() {
List<CompletableFuture<T>> pendingFutures = futures.stream().filter(future -> !future.isDone()).toList();
log.trace("Completing operation for {} futures, expired {}", futures.size(), pendingFutures.size());
pendingFutures.forEach(future -> future.completeExceptionally(new TimeoutException("Request has been timed out after " + timeoutMs + " ms")));
responseCallback.run();
}
/**
* This is invoked after onComplete(), so no actions required.
*/
@Override
public void onExpiration() {
// This is invoked after onComplete(), so no actions required.
}
}
Run 1
The code is well-structured with clear separation of concerns between tryComplete(), onComplete(), and onExpiration(). The class has a single, focused responsibility. Nesting is minimal - only one level in the stream filter operations. Intent is clearly documented with meaningful method names and helpful Javadoc. The code is concise without being cryptic. Action granularity is mostly good, though the line calling completeExceptionally with inline exception construction is somewhat long. Reading flow is natural - the tryComplete/onComplete pattern follows Kafka's DelayedOperation conventions. Minor improvement could be extracting the pending futures calculation to avoid duplication between methods. The constant 0x0 is slightly unusual instead of just 0, but doesn't hurt readability significantly. Overall, this is clean, maintainable code that clearly expresses its purpose.
Run 2
The code is well-structured and follows a clear logical organization with appropriate class-level Javadoc and method-level comments. Nesting is minimal, with a simple if-else in tryComplete and a stream pipeline in onComplete. Intent is clearly explained through documentation: the class manages delayed operations using CompletableFutures for ACL updates. Code length is appropriate—concise without sacrificing clarity. Action granularity is mostly good, though the stream-filter-toList line in onComplete combines multiple operations and is somewhat long. Reading flow is natural: tryComplete checks completion, onComplete handles timeout/finalization, and onExpiration explicitly documents a no-op. Minor issues: the hex literal 0x0 instead of 0 is unnecessarily cryptic, and the single-line trace statement in onComplete is long. Overall, very readable code with clear purpose and good structure.
Run 3
The code is well-structured and highly readable. It has a clear package and import organization, a concise class Javadoc explaining its purpose and usage context, and methods that follow a logical, flat flow with minimal nesting. Each method performs a single, well-defined action—tryComplete checks pending futures, onComplete handles timeout and invokes the callback, and onExpiration is intentionally empty. Variable names like 'pendingFutures' and 'responseCallback' convey intent clearly. The use of streams and lambda expressions is appropriate and does not sacrifice clarity. Method and field documentation further improve understanding. The code length is proportionate to its complexity, with no redundant code. The only minor observation is the use of a hexadecimal literal (0x0) instead of a plain 0, which is slightly unusual but not confusing. Overall, the code maintains a natural reading flow and would be easy for a developer to understand at a glance.
Run 1
Code Structure: Fields, constructor, and lifecycle methods are logically organized, with responsibilities clearly separated. Nesting: Control flow is shallow, limited to a simple if/else and short lambdas. Clarity of intent: Descriptive names and documentation clearly explain completion checks, timeout handling, and callback invocation; the hexadecimal zero (0x0) is unnecessarily unconventional. Code length: The implementation is concise, though the else after a return and duplicated onExpiration explanation add minor redundancy. Action granularity: Most statements perform one clear action, but the long forEach statement combines exception creation and future completion. Reading flow: The lifecycle methods follow a natural sequence, supported by useful logging; long stream expressions and minimal spacing around imports slightly reduce scanability.
Run 2
Code Structure: Fields, constructor, and lifecycle methods are logically organized with clear responsibilities. Nesting: Minimal nesting; only a simple if/else and short lambdas. Clarity of intent: Descriptive names, documentation, and logging explain completion and timeout behavior, though hexadecimal 0x0 is unnecessarily unconventional. Code length: Compact overall, with minor redundancy in the empty-method comment and an else after an early return. Action granularity: Statements generally perform one action each; the exceptional-completion lambda combines exception construction and future completion on a long line. Reading flow: Natural progression from initialization to completion checks, completion handling, and expiration; long stream and lambda lines slightly hinder scanning.
Run 3
Code Structure: Fields, constructor, and lifecycle methods are logically organized with clear separation of responsibilities. Nesting: Minimal nesting; tryComplete uses a simple conditional, and stream operations keep iteration flat. Clarity of intent: Descriptive names, documentation, and logging explain completion and timeout behavior, though hexadecimal 0x0 is unnecessarily unconventional. Code length: Compact overall, with minor redundancy in the empty onExpiration method's documentation and comment. Action granularity: Most lines perform one clear action; the long completeExceptionally lambda combines exception construction and future completion, making it slightly harder to scan. Reading flow: The progression from initialization to completion checking, completion handling, and expiration is natural. Long stream/lambda lines and tightly packed imports slightly reduce readability.