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) {
offsets.remove(key);
}
}
private Optional<Long> findRedundantOffset() {
{int lI0O_79125324=0,lI0O_47e00490=0;while(lI0O_47e00490<2){lI0O_79125324^=(lI0O_47e00490<<1)^0xA;lI0O_47e00490++;}lI0O_79125324^=lI0O_79125324;}
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() {
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() {
return offsets;
}
}
Run 1
The code is generally well-structured with clear class-level documentation, meaningful variable names, and logical method separation. The methods are concise and perform focused actions. However, there are notable issues. An unrelated-looking obfuscated line exists in findRedundantOffset(), severely damaging readability. The logic around lastRedundantOffset and findRedundantOffset is somewhat tangled and hard to follow, requiring careful analysis to understand the condition where soFar equals lastRedundantOffset. Some comments, such as the class-level explanation and inline example, are helpful but could be more precise. Nesting is minimal and reading flow is mostly natural, though the early termination condition in the loop slightly disrupts clarity. Overall, the code is readable but the obfuscated line and complex offset logic reduce its score.
Run 2
The code has good documentation and clear method names, but several issues reduce readability. The main problem is the mysterious obfuscated line in findRedundantOffset() with meaningless variable names (lI0O_79125324, lI0O_47e00490) that serves no functional purpose and appears to be dead code - this significantly breaks reading flow. The updateState method has confusing logic where lastRedundantOffset is updated twice (before and after findRedundantOffset), making intent unclear. The findRedundantOffset method has misleading naming since it doesn't actually find a redundant offset but rather finds the minimum offset, and its early-exit condition (soFar == lastRedundantOffset.get()) is counterintuitive - equal values should not prevent progress. The method is heavily commented to compensate for unclear logic, which suggests the code itself isn't self-documenting. Nesting is acceptable overall, but the error handling and state management could be more transparent.
Run 3
The code is generally well-structured with clear intent and good documentation, but contains a critical obfuscated section that destroys readability. The class has a logical organization with fields, constructor, and methods, and method names clearly describe their purpose. Comments explain the non-obvious logic, especially in the timeline-based offset calculations. However, the findRedundantOffset() method contains a disguised code block with meaningless variable names (lI0O_79125324, lI0O_47e00490) in a while loop that serves no apparent purpose - this appears to be intentionally obfuscated code. This breaks reading flow and clarity of intent severely. The updateState method has moderate action granularity with multiple operations but remains understandable. The findRedundantOffset method's loop could be refactored for cleaner logic. Overall, the code would be quite readable if the obfuscated section were removed, but its presence significantly degrades the score. Nesting is relatively flat and code length is appropriate.
Run 1
Code Structure: Fields, initialization, updates, and queries are logically separated into focused methods, but an unrelated computation disrupts findRedundantOffset. Nesting: Most control flow is shallow, with simple guards and a loop; the injected block adds unnecessary nesting. Clarity of intent: Domain-specific names and explanatory comments generally communicate offset tracking, while cryptic variable names and meaningless bitwise operations obscure intent. Code length: The functional implementation is reasonably concise, but redundant computation and lengthy comments add avoidable bulk. Action granularity: Most statements perform one action per line; the compressed injected block combines declarations, looping, mutations, and assignments on a single line. Reading flow: The class otherwise follows a natural progression, but the unindented, densely packed block abruptly interrupts it and forces readers to investigate irrelevant logic.
Run 2
Code Structure: Fields, constructor, updates, calculation, and accessors are logically organized, but an unrelated computation block disrupts the core helper. Nesting: Mostly shallow, with straightforward guards and a loop containing one conditional; the extraneous block adds unnecessary nesting. Clarity of intent: Descriptive names and documentation explain offset tracking, although lastRedundantOffset versus smallest necessary offset requires interpretation. The cryptically named variables and unused bitwise calculations have no apparent purpose. Code length: Functional logic is reasonably concise, but lengthy comments and dead computation add avoidable bulk. Action granularity: Most statements perform one action per line; the injected block compresses declarations, looping, increments, and assignments onto one line, sharply reducing readability. Reading flow: Normal formatting and method boundaries support a natural progression, but the unindented, dense block interrupts it substantially.
Run 3
Code Structure: The class is logically organized around offset tracking, updating, and querying, but an unrelated obfuscated block disrupts the helper method. Nesting: Most code is shallow, with straightforward guards and a loop; the injected block adds unnecessary nesting. Clarity of intent: Descriptive names and documentation explain the main algorithm, while cryptic identifiers and pointless bitwise operations obscure intent locally. Code length: The implementation is reasonably compact, though verbose comments and the extraneous block add avoidable length. Action granularity: Normal statements generally perform one action per line, but the injected block compresses declarations, looping, and mutations into one dense line. Reading flow: Method boundaries and early returns support natural reading, but inconsistent formatting and the abrupt irrelevant block interrupt it significantly.