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 == 0))) {
log.trace("{} future still pending, not completing the delayed operation", pending);
return false;
} else {
log.trace("All futures have been completed or have errors, completing the delayed operation");
return forceComplete();
}
}
/**
* 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 and readable. It uses clear class and method names, concise comments explaining purpose, and follows a logical flow. The tryComplete method is slightly verbose with an unnecessary if-else and a redundant double negation, but it remains clear. Nesting is minimal, and each action is on its own line, aiding readability. The use of streams and lambda expressions is appropriate and concise. Overall, the code is easy to follow and maintain, with only minor stylistic improvements possible.
Run 2
The code is generally readable and well-structured. It follows a clear class hierarchy with a single responsibility. Naming is meaningful (e.g., tryComplete, onComplete, futures). The class is concise and avoids unnecessary verbosity. However, readability is slightly reduced by a few issues: the condition `if (!((pending == 0)))` is awkward and could be simplified to `if (pending != 0)`. The logging messages are somewhat repetitive, and the `onComplete` method performs multiple logical actions (filtering, logging, timing out futures, invoking callback) with minimal separation, which slightly hinders action granularity. The `onExpiration` method has an empty body with just a comment, which is acceptable but adds minor noise. Overall, the intent is clear, the code is flat, and the reading flow is natural, but minor syntactic and structural cleanups would improve clarity.
Run 3
The code is well-structured with clear separation of concerns and logical organization. Import statements are minimal and relevant. Nesting is flat, with no deeply nested blocks. The class and method purposes are documented with clear Javadoc, and method names are descriptive (tryComplete, onComplete, onExpiration). The code length is appropriate, with concise methods that avoid unnecessary verbosity. Each line in the lambda expressions and stream operations performs a single action, maintaining good action granularity. The reading flow is natural - the tryComplete method checks completion status, then onComplete handles timeout/finalization logically. Minor improvements possible: the condition `!((pending == 0))` could be simplified to `pending != 0`, and the empty onExpiration method could be justified with a comment explaining the override is required by the abstract method pattern.
Run 1
Code Structure: Well-organized class with clearly separated initialization, completion checking, completion handling, and expiration handling. Nesting: Shallow, though the else after an early return is unnecessary. Clarity of intent: Descriptive names and method documentation explain the lifecycle; the negated, redundantly parenthesized condition !((pending == 0)) is less clear than pending > 0. Code length: Generally concise, with minor redundancy in the expiration comment and conditional structure. Action granularity: Mostly one action per line; the long completeExceptionally lambda combines exception construction and completion, making it harder to scan. Reading flow: Methods follow a natural lifecycle order, and logging supports comprehension. Long stream expressions and minimal separation around imports slightly reduce readability.
Run 2
Code Structure: Fields, constructor, and lifecycle methods are logically organized with focused responsibilities. Nesting: Shallow nesting keeps control flow easy to follow, although the else after an early return is unnecessary. Clarity of intent: Descriptive names, logging, and documentation explain completion and timeout behavior; the negated, redundantly parenthesized condition !((pending == 0)) is less clear than pending != 0. Code length: Generally concise, with minor redundancy in comments and conditional syntax. Action granularity: Statements largely perform one action each; stream pipelines are cohesive, though the long exceptional-completion lambda would benefit from line wrapping. Reading flow: The progression from initialization through completion checks, completion handling, and expiration is natural. Dense stream lines and tightly grouped imports slightly reduce scanability.
Run 3
Code Structure: Fields, constructor, and lifecycle methods are logically organized, with focused responsibilities. Nesting: Shallow, although the else after an early return is unnecessary. Clarity of intent: Descriptive names, documentation, and logging make completion and timeout behavior clear; the negated, over-parenthesized condition obscures a simple pending != 0 check. Code length: Generally concise, but repeated documentation in onExpiration adds redundancy. Action granularity: Statements mostly perform one action each; stream expressions compactly represent related operations, though the long exceptional-completion lambda is harder to scan. Reading flow: Natural progression from initialization to completion checks and callbacks, slightly weakened by tightly grouped imports and long lines.