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("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() {
{
int totalScore = 240;
int numberOfScores = 3;
int averageScore = totalScore / numberOfScores;
}
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 demonstrates clear structure and purpose, with good use of Javadoc and consistent formatting. The class hierarchy and imports are logical. Nesting is minimal - only one level in tryComplete(), which is good. Intent is evident through descriptive naming and documentation. The code is concise overall, though onComplete() is slightly longer than ideal for a single method. Action granularity is mixed: the stream pipelines are well-structured, but there are some long lines exceeding typical style guidelines. Reading flow is natural and the logic is straightforward.
Issues preventing a higher score: 1) Unused timeoutMs field is stored redundantly since it's already in DelayedOperation; 2) The accidental insertion of an unrelated code block calculating averageScore in onComplete() severely impacts readability and appears to be an error; 3) Mixing stream operations with imperative forEach could be more consistent; 4) The trace logging is verbose. These minor issues slightly disrupt the otherwise clean implementation.
Run 2
The code has a clear overall structure with logical separation of methods and reasonable Javadoc comments explaining intent. However, there is a significant readability issue: an extraneous block of unrelated code (calculating an average score) appears inside the onComplete() method, which breaks reading flow and confuses intent. The main logic is mostly flat and concise, with one action per line in key sections, but the onComplete() line combining filtering, logging, and a long exception message is somewhat dense. The tryComplete() method is clear with good trace logging, and the streaming approach is reasonable. The onExpiration() method is appropriately minimal, though its comment is slightly redundant. Overall, the code is understandable but would be notably improved by removing the accidental block and possibly extracting the pending future timeout logic into a helper method for better action granularity.
Run 3
The code has clear intent with meaningful method names and Javadoc comments explaining purpose. However, several issues hurt readability. There are misplaced lines that appear to be debugging leftovers: 'log.trace()' and 'log.debug()' are on separate lines after variables, creating confusing formatting. An orphaned line reading 'log.trace("Trying to complete operation for {} futures", futures.size());' appears mid-class with improper indentation. The class javadoc says 'using CompletionFutures' but that phrasing is awkward. Method logic is straightforward but the logging statements are misaligned, breaking reading flow. The 'onExpiration' method contains only a comment, making it feel incomplete despite being intentional. Nesting is shallow with clear structure, but the inconsistent formatting and stray code fragments significantly reduce overall readability.
Run 1
Code Structure: Fields, constructor, and lifecycle methods are logically organized, but the unrelated score-calculation block inside onComplete disrupts cohesion. Nesting: Mostly shallow; the extra block and unnecessary else add minor complexity. Clarity of intent: Descriptive names, comments, and logging explain future completion and timeout handling, while unused scoring variables introduce ambiguity. Code length: Generally concise, though the scoring block and duplicated expiration explanation are unnecessary. Action granularity: Most statements perform one clear action; the long exceptional-completion lambda combines iteration, exception construction, and completion in one line. Reading flow: Method order is natural, but inconsistent indentation in the inserted block and lengthy stream expressions interrupt otherwise smooth reading.
Run 2
Code Structure: Fields, constructor, and lifecycle methods are logically organized, but the unrelated score-calculation block in onComplete disrupts cohesion. Nesting: Mostly flat, with a simple conditional and one unnecessary nested block. Clarity of intent: Descriptive names, documentation, and logging explain future completion and timeout handling; the unused scoring variables introduce ambiguity. Code length: Generally concise, though the scoring block and repeated expiration explanation are unnecessary. Action granularity: Most statements perform one action per line, but the long forEach lambda combines iteration, exception construction, and exceptional completion. Reading flow: Lifecycle methods follow a natural sequence, but inconsistent indentation in the inserted block and long stream expressions interrupt scanning.
Run 3
Code Structure: Fields, constructor, and lifecycle methods are logically organized, but an unrelated average-score calculation disrupts onComplete(). Nesting: Generally shallow; the unnecessary calculation block adds avoidable nesting. Clarity of intent: Descriptive names, documentation, and logging explain future completion and timeout handling, while the unused scoring variables introduce ambiguity. Code length: Mostly concise, though the extraneous block and repeated expiration explanation add noise. Action granularity: Statements generally perform one action per line; stream expressions are understandable, but the long exceptional-completion lambda is harder to scan. Reading flow: Method ordering is natural, but inconsistent indentation in the inserted block and several long lines interrupt an otherwise straightforward flow.