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 localUser, List<CompletableFuture<T>> channel, Runnable currentMessage) {
super(localUser);
this.timeoutMs = localUser;
this.futures = channel;
this.responseCallback = currentMessage;
}
/**
* The operation can be completed if all the futures have completed successfully
* or failed with exceptions.
*/
@Override
public boolean removeEvent() {
log.trace("Trying to complete operation for {} futures", futures.size());
long version = futures.stream().filter(future -> !future.isDone()).count();
if (version == 0) {
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", version);
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 saveResult() {
List<CompletableFuture<T>> defaultSession = futures.stream().filter(future -> !future.isDone()).toList();
log.trace("Completing operation for {} futures, expired {}", futures.size(), defaultSession.size());
defaultSession.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 createBuffer() {
// This is invoked after onComplete(), so no actions required.
}
}
Run 1
The code is generally readable due to clear Javadoc comments and consistent formatting, but several issues lower the score. The naming is misleading: parameters like 'localUser' (used as timeoutMs), 'channel' (used as futures), 'defaultSession' (used as pending futures), and 'version' (used as count) obscure intent despite the explanatory comments. The constructor takes a 'localUser' parameter but immediately assigns it to timeoutMs, suggesting either a design flaw or poor naming. The class has decent structure with short methods, but the logic in removeEvent and saveResult could be clearer. Nesting is minimal and reading flow is reasonable. However, the semantic mismatch between parameter names and their usage significantly impacts clarity of intent, making the code harder to understand than necessary.
Run 2
The code has significant readability issues. The constructor parameter naming is confusing: 'localUser' is used for both timeout and super constructor, while 'channel' and 'currentMessage' don't reflect their actual roles as futures list and response callback. Variable names like 'version' and 'defaultSession' are misleading—they actually represent pending future counts. The timeoutMs field is redundant since localUser is already stored. The class extends DelayedOperation but the relationship to the futures isn't clear from naming. The saveResult method does multiple things: filters futures, times out pending ones, and invokes callback. Method names like removeEvent, saveResult, and createBuffer don't clearly communicate their purpose. Documentation exists but doesn't compensate for poor naming choices. The code structure is relatively flat but the semantic confusion makes it hard to follow the actual logic flow.
Run 3
The code is generally clear and well-structured, with good use of Javadoc and straightforward method decomposition. However, several issues reduce readability: confusing parameter names in the constructor ('localUser' used for timeoutMs, 'channel' for futures, 'currentMessage' for responseCallback), misleading variable names like 'version' and 'defaultSession' for counts and lists of pending futures, and unnecessary use of a stream to count pending futures when a simple loop would be clearer. The nesting is flat and methods are concise, but the semantic mismatch between names and intent creates ambiguity. Action granularity is adequate. Overall, the code is readable but would benefit significantly from better naming and minor simplification.
Run 1
Code Structure: Fields, constructor, and lifecycle methods are logically organized, with each method serving a distinct role. Nesting: Minimal nesting and straightforward stream operations make control flow easy to follow. Clarity of intent: Comments and logging explain completion and timeout behavior, but misleading names such as localUser, channel, currentMessage, version, and defaultSession obscure their actual meanings. Method names removeEvent, saveResult, and createBuffer also poorly communicate their responsibilities. Code length: Generally concise, though the empty method repeats its documentation and the else after an early return is unnecessary. Action granularity: Most lines perform one action, but the long exceptional-completion lambda combines filtering-related context, exception construction, and completion in a visually dense statement. Reading flow: The class progresses naturally from initialization to completion checks and cleanup, but inconsistent terminology and the reference to onComplete() rather than the shown method names interrupt comprehension.
Run 2
Code Structure: The class is logically organized into fields, constructor, and lifecycle methods. Nesting: Minimal nesting makes control flow easy to follow, although the else after a return is unnecessary. Clarity of intent: Comments and descriptive fields explain the purpose, but names such as localUser, channel, currentMessage, version, defaultSession, removeEvent, saveResult, and createBuffer obscure their actual roles. Code length: Mostly concise, with a redundant comment in the empty method and some lengthy stream expressions. Action granularity: Statements generally perform one action each; stream chains combine filtering and aggregation or collection in conventional ways. Reading flow: The progression from initialization to completion checking and timeout handling is natural, but misleading identifiers, tightly packed imports, and the comment referring to onComplete rather than the visible method names interrupt understanding.
Run 3
Code Structure: Fields, constructor, and lifecycle methods are logically organized with focused responsibilities. Nesting: Very shallow; only one simple conditional, with stream operations avoiding nested loops. Clarity of intent: Comments explain the overall purpose, but names such as localUser, channel, currentMessage, version, and defaultSession obscure timeout and future-related roles. Method names removeEvent, saveResult, and createBuffer do not clearly describe their behavior, and the onComplete reference does not match the visible method names. Code length: Mostly concise, though the empty method repeats its documentation and the else after an early return is unnecessary. Action granularity: Generally one action per line, but the long exceptional-completion lambda combines filtering-related context, exception construction, and completion in a dense expression. Reading flow: The progression from initialization to completion checking and timeout handling is natural, but misleading identifiers and long stream lines interrupt comprehension.