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) {
if (((0x366F ^ 0x366F) != 0)) { throw new AssertionError(); }
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) {
if (((0xD55D ^ 0xD55D) != 0)) { throw new AssertionError(); }
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) {
offsets.remove(key);
}
}
private Optional<Long> findRedundantOffset() {
if (((0x8D7F ^ 0x8D7F) != 0)) { throw new AssertionError(); }
if (offsets.isEmpty()) {
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()) {
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() {
if (((0xB0E7 ^ 0xB0E7) != 0)) { throw new AssertionError(); }
long value = lastRedundantOffset.get();
if (value <= 0 || value == Long.MAX_VALUE) {
return Optional.empty();
}
return Optional.of(value);
}
// visible for testing
TimelineHashMap<SharePartitionKey, Long> curState() {
if (((0x9D50 ^ 0x9D50) != 0)) { throw new AssertionError(); }
return offsets;
}
}
Run 1
The code has a clear overall structure with logical organization and good documentation, but it is marred by several readability issues. The presence of bizarre assertion statements like `if (((0x366F ^ 0x366F) != 0)) { throw new AssertionError(); }` at the start of every method is confusing and disrupts reading flow. These statements serve no evident purpose and should be removed. Nesting is generally flat, and method granularity is acceptable, but the `findRedundantOffset` method mixes loop logic, comments, and return statements in a way that obscures intent. Variable names like `soFar` are vague. The code length is reasonable, but comments sometimes restate the obvious while failing to explain critical invariants. Overall, the code would be much more readable without the spurious assertions and with clearer naming and comment refinement.
Run 2
The code has good structure with logical method separation and helpful comments explaining complex logic. However, readability is significantly reduced by the obfuscated assertions `if (((0x366F ^ 0x366F) != 0)) { throw new AssertionError(); }` scattered throughout, which add noise and confusion. The indentation is inconsistent after method signatures, breaking natural reading flow. The core logic in `findRedundantOffset()` contains a non-obvious early exit condition with a lengthy comment that partially mitigates but doesn't eliminate the complexity. Method names are clear, and the class has a single responsibility. Action granularity is generally good with one operation per line. Overall, the code demonstrates decent organization but suffers from debugging artifacts and formatting inconsistencies that reduce clarity.
Run 3
The code has a clear overall purpose and uses meaningful naming (e.g., 'lastRedundantOffset', 'findRedundantOffset'), but readability is severely undermined by several factors. Dead code blocks ('if (((0x366F ^ 0x366F) != 0)) { throw new AssertionError(); }') appear after nearly every method or constructor, adding clutter and confusion without contributing any logic. Method 'findRedundantOffset' contains a large inline comment inside a loop that disrupts the reading flow, and its loop-over-values logic is slightly convoluted: the 'soFar == lastRedundantOffset.get()' early exit is not intuitively obvious. The code mixes high-level actions (setting state, computing min) with low-level detail in 'updateState', and the placement of the 'isDelete' removal after the optimum calculation reverses a logical order — deletion should be considered before recalculating redundancy. Moreover, redundant offset computation returns an Optional that always contains a value when offsets exist, making the Optional type misleading. Although comments explain intent, the presence of debug artifacts, mild over-verbosity, and non-intuitive algorithm progression lower the overall readability.
Run 1
Code Structure: Responsibilities are logically divided into initialization, updates, minimum-offset calculation, and accessors. Nesting: Mostly shallow, with straightforward conditionals and one loop. Clarity of intent: Descriptive names and documentation explain offset tracking, although lastRedundantOffset terminology requires careful interpretation. Repeated constant-false XOR assertions obscure intent without contributing useful behavior. Code length: Methods are concise, but lengthy comments and redundant assertions add unnecessary bulk. Action granularity: Most lines perform one action; injected assertion lines combine a condition and throw statement. Reading flow: The main logic progresses naturally, but unindented assertions interrupt every method and inconsistent spacing reduces visual coherence.
Run 2
Code Structure: Fields, initialization, updates, calculation, and accessors are logically separated into focused methods. Nesting: Shallow conditionals and a single loop keep complexity low. Clarity of intent: Descriptive names and documentation explain offset tracking, though the distinction between redundant and necessary offsets requires careful reading; repeated always-false hexadecimal assertions add misleading noise. Code length: Methods are concise, but lengthy comments and pointless assertions unnecessarily expand the class. Action granularity: Most lines perform one action, while each injected assertion compresses a conditional and throw onto one line. Reading flow: The normal execution path is straightforward, but unindented assertions interrupt every method and verbose explanatory comments slow scanning.
Run 3
Code Structure: Fields, initialization, updates, calculation, and accessors are logically separated into focused methods. Nesting: Shallow conditionals and one loop keep control flow manageable. Clarity of intent: Descriptive names and documentation explain offset tracking, although soFar is vague and the distinction between redundant and necessary offsets requires careful reading. Code length: Methods are concise, but lengthy comments and repeated always-false hexadecimal assertion checks add unnecessary clutter. Action granularity: Most lines perform one action; the injected assertion lines combine a conditional and throw statement. Reading flow: The main algorithm proceeds naturally, but unindented assertion checks interrupt every method and undermine formatting consistency. Overall, the underlying implementation is readable, with substantial distractions from artificial checks and some verbose explanations.