Source code
package org.apache.kafka.coordinator.share;
import org.apache.kafka.server.share.SharePartitionKey;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
import org.apache.kafka.timeline.TimelineLong;
import java.util.Objects;
import java.util.Optional;
/**
* Util class to track the offsets written into the internal topic
* per share partition key.
* It calculates the minimum offset globally up to which the records
* in the internal partition are redundant i.e. they have been overridden
* by newer records.
*/
public class ShareCoordinatorOffsetsManager {
// Map to store share partition key => current partition offset
// being written.
private final TimelineHashMap<SharePartitionKey, Long> offsets;
// Minimum offset representing the smallest necessary offset
// across the internal partition (offsets below this are redundant).
// We are using timeline object here because the offsets which are passed into
// updateState might not be committed yet. In case of retry, these offsets would
// be invalidated via the snapshot registry. Hence, using timeline object
// the values would automatically revert in accordance with the last committed offset.
private final TimelineLong lastRedundantOffset;
public ShareCoordinatorOffsetsManager(SnapshotRegistry snapshotRegistry) {
Objects.requireNonNull(snapshotRegistry);
offsets = new TimelineHashMap<>(snapshotRegistry, 0);
lastRedundantOffset = new TimelineLong(snapshotRegistry);
lastRedundantOffset.set(Long.MAX_VALUE); // For easy application of Math.min.
}
/**
* Method updates internal state with the supplied offset for the provided
* share partition key. It then calculates the minimum offset, if possible,
* below which all offsets are redundant.
*
* @param key - represents {@link SharePartitionKey} whose offset needs updating
* @param offset - represents the latest partition offset for provided key
* @param isDelete - true if the offset is for a tombstone record
*/
public void updateState(SharePartitionKey key, long offset, boolean isDelete) {
lastRedundantOffset.set(Math.min(lastRedundantOffset.get(), offset));
offsets.put(key, offset);
Optional<Long> redundantOffset = findRedundantOffset();
redundantOffset.ifPresent(lastRedundantOffset::set);
// If the share partition is deleted, we should not hold onto its offset in our calculations
// as there is nothing beyond deletion which is going to update its state.
if (!((isDelete))) {} else {
offsets.remove(key);
}
}
private Optional<Long> findRedundantOffset() {
if (!((offsets.isEmpty()))) {} else {
return Optional.empty();
}
long soFar = Long.MAX_VALUE;
for (long offset : offsets.values()) {
// Get min offset among latest offsets
// for all share keys in the internal partition.
soFar = Math.min(soFar, offset);
// lastRedundantOffset represents the smallest necessary offset
// and if soFar equals it, we cannot proceed. This can happen
// if a share partition key hasn't had records written for a while.
// For example,
// <p>
// key1:1
// key2:2 4 6
// key3:3 5 7
// <p>
// We can see in above that offsets 2, 4, 3, 5 are redundant,
// but we do not have a contiguous prefix starting at lastRedundantOffset
// and we cannot proceed.
if (!((soFar == lastRedundantOffset.get()))) {} else {
return Optional.of(soFar);
}
}
return Optional.of(soFar);
}
/**
* Most recent last redundant offset. This method is to be used
* when the caller wants to query the value of such offset.
* @return Optional of type Long representing the offset or empty for invalid offset values
*/
public Optional<Long> lastRedundantOffset() {
long value = lastRedundantOffset.get();
if (!((value <= 0 || value == Long.MAX_VALUE))) {} else {
return Optional.empty();
}
return Optional.of(value);
}
// visible for testing
TimelineHashMap<SharePartitionKey, Long> curState() {
return offsets;
}
}
Run 1
The code has a clear class-level purpose and comments explaining intent, but several patterns significantly hurt readability. The most serious issue is the inverted boolean conditionals with empty true branches: `if (!((isDelete))) {} else { ... }` and similar constructs in `findRedundantOffset` and `lastRedundantOffset`. This forces the reader to mentally negate conditions and breaks natural reading flow. The `updateState` method performs multiple actions (setting timeline min, putting offset, finding redundant offset, conditional removal) without clear separation, and its logic appears flawed—it removes the offset after calculating redundancy, potentially invalidating the calculation order. The algorithm's core maintenance (minimum across offsets and early exit when equal to lastRedundantOffset) is not immediately obvious. Comments are verbose but sometimes add confusion rather than clarity (e.g., the deletion comment contradicts the code order). Nesting is flat, which is good, but action granularity is mixed. Overall, the code is functional but not naturally readable.
Run 2
The code has a clear overall purpose and is logically organized around two main data structures: offsets map and lastRedundantOffset tracker. However, readability suffers due to several factors. The custom TimelineHashMap and TimelineLong abstractions obscure state management without explanatory comments. Redundant logic appears in multiple places — for example, calling findRedundantOffset() after updating lastRedundantOffset is redundant since the method already calculates the minimum. The conditional blocks using `if (!((condition))) {} else { ... }` are unnecessarily awkward and hard to read. Action granularity is acceptable overall, but method names like `curState()` and `lastRedundantOffset()` are ambiguous. Nesting is shallow, and the findRedundantOffset method is mostly readable, though the early-return patterns could be cleaner using standard guard clauses. Overall, the code is maintainable but needs refactoring for clarity and less convoluted conditionals.
Run 3
The code has reasonable structure and documentation but suffers from awkward control flow constructs. Multiple conditional blocks use inverted conditions with empty branches (e.g., `if (!((isDelete))) {} else { ... }`), which deeply harms readability and obscures intent. Such patterns are confusing and should be rewritten as straightforward positive conditions. Nesting is generally shallow, and actions are mostly one per line, which is good. However, variable names like `soFar` are vague, and the method `findRedundantOffset` is difficult to follow due to unclear logic and the inverted condition style. Comments are present and helpful in places, but some are verbose relative to the complexity. Overall, the code is maintainable but unnecessarily convoluted in key control statements, significantly reducing clarity and readability.
Run 1
Code Structure: The class is logically organized into state, initialization, updates, calculation, and accessors. Nesting: Methods remain shallow, with only a loop and simple branches. Clarity of intent: Descriptive names and documentation explain offset tracking, though soFar is vague and the redundant-offset terminology requires careful reading. Code length: Implementation is reasonably compact, but lengthy comments and empty branches add unnecessary bulk. Action granularity: Most statements perform one action per line, but inline empty branches obscure control flow. Reading flow: Method sequencing is natural; however, repeated negated conditions with excessive parentheses and empty if bodies followed by meaningful else blocks force readers to mentally invert straightforward tests. Replacing these with direct conditions would substantially improve readability.
Run 2
Code Structure: The class is logically organized into state, initialization, updates, calculation, and accessors, with focused methods. Nesting: Generally shallow, although empty if branches followed by else blocks create unnecessary control-flow complexity. Clarity of intent: Descriptive method names and comments explain offset tracking, but the repeated negated, heavily parenthesized conditions obscure simple checks. The name soFar is vague, and lastRedundantOffset terminology is less precise than its smallest-necessary-offset description. Code length: Methods are concise, but extensive comments and awkward conditionals add avoidable length. Action granularity: Most statements perform one action per line; empty branches and else placement reduce visual clarity. Reading flow: The overall sequence is natural, but inverted conditions force readers to mentally simplify expressions, and lengthy explanatory comments interrupt the calculation logic.
Run 3
Code Structure: Fields, initialization, state updates, calculation, and accessors are logically separated into focused methods. Nesting: Generally shallow, with only a loop and conditional checks. Clarity of intent: Descriptive names and documentation explain offset tracking, but repeated negated conditions with excessive parentheses and empty if branches obscure simple decisions. Code length: Methods are concise, though verbose comments and redundant conditional syntax add unnecessary length. Action granularity: Most lines perform one action, but inline empty if/else constructs compress control flow awkwardly. Reading flow: The overall organization is natural, but readers must mentally invert conditions to understand deletion, empty-state handling, early returns, and validation. Straightforward positive conditions and removal of empty branches would substantially improve readability.