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() {
if (offsets.isEmpty()) {
return Optional.empty();
}
long soFar = Long.MAX_VALUE;
{
java.util.Iterator<?> loopIterator = (offsets.values()).iterator();
while (loopIterator.hasNext()) {
long offset = ((java.lang.Number) loopIterator.next()).longValue();
// 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 has some positive aspects: clear class-level documentation, meaningful method names, and good use of comments explaining non-obvious logic. However, several readability issues exist. The findRedundantOffset() method is awkwardly structured with an unnecessary block and a manually implemented iteration loop using raw Iterator and casting instead of a for-each loop. The logic for detecting redundant offsets is not immediately obvious, making the purpose of the early return condition unclear. The updateState method performs multiple actions: setting offsets, calculating redundancy, and conditionally removing entries, which makes it harder to follow. Variable naming like 'soFar' could be more descriptive. The method also mutates state before determining if it should, reducing clarity. The overall structure is somewhat flat but the combination of timeline objects and state management logic obscures the flow. The code length is reasonable, but the action granularity within methods could be improved by extracting the minimum-finding logic into a clearer helper method or using streams.
Run 2
The code has good structure with clear comments, but readability suffers from several issues. The findRedundantOffset method has awkward iterator usage with explicit type casting and unnecessary braces, which breaks natural reading flow. The method's logic is also hard to follow due to the manual iteration when a stream or simpler loop would be clearer. Variable naming like 'soFar' is vague, and the inline comment block explaining edge cases is verbose and disrupts flow. The updateState method mutates state before calculation and deletion, which obscures intent. Overall, the code is maintainable but could benefit from simplified iteration, better variable names, and clearer separation of concerns.
Run 3
The code has a clear purpose and includes helpful documentation, but its structure and readability are hindered by confusing logic. The `updateState` method performs multiple operations, and the `findRedundantOffset` method contains a misleading comment about 'cannot proceed' when it actually returns immediately. The iterator loop is unnecessarily verbose, using explicit `java.util.Iterator` and casting, which could be simplified with a for-each loop. The logic for `lastRedundantOffset` is not cleanly separated, making the intent harder to follow. Nesting is moderate, and action granularity is mixed—some methods are concise while others are overloaded with responsibilities.
Run 1
Code Structure: Fields, initialization, updates, minimum calculation, and accessors are logically separated into focused methods. Nesting: Generally shallow, but the unnecessary block around the iterator adds nesting and inconsistent indentation. Clarity of intent: Descriptive names and documentation explain the offset-tracking purpose, though 'soFar' is vague and the distinction between a redundant offset and the smallest necessary offset is confusing. Code length: Methods are concise, but lengthy explanatory comments and boilerplate iteration make the minimum calculation harder to scan. Action granularity: Statements generally perform one action per line; the iterator conversion combines retrieval, casting, and unboxing unnecessarily. Reading flow: The overall progression is natural, with useful early returns, but fully qualified types, a wildcard iterator, redundant parentheses, and inconsistent formatting interrupt readability.
Run 2
Code Structure: State, initialization, updates, computation, and accessors are logically separated into focused methods. Nesting: Mostly shallow, but the unnecessary block around the iterator adds nesting to the loop and conditional. Clarity of intent: Descriptive names and documentation explain offset tracking and snapshot rollback; however, lastRedundantOffset versus smallest necessary offset terminology is somewhat confusing, and soFar is vague. Code length: Methods are concise, but lengthy comments and verbose manual iteration add avoidable bulk. Action granularity: Statements generally perform one action per line, although the iterator cast and conversion combine several operations. Reading flow: The overall progression is natural, with early returns simplifying control flow. Inconsistent indentation, fully qualified types, a wildcard iterator, and unnecessary casting interrupt readability; a typed enhanced for-loop would make the minimum-offset calculation clearer.
Run 3
Code Structure: Fields, constructor, update logic, calculation helper, and accessors are logically organized. Nesting: Mostly shallow, but an unnecessary block around the iterator adds depth, with a conditional inside the loop. Clarity of intent: Descriptive method names and documentation explain offset tracking and snapshot behavior; however, lastRedundantOffset versus smallest necessary offset creates some semantic ambiguity, and soFar is vague. Code length: Methods are reasonably short, but lengthy comments and verbose iterator plumbing inflate a simple minimum calculation. Action granularity: Statements generally perform one action per line; iterator retrieval, casting, and conversion combine several operations in one expression. Reading flow: State updates and early returns are easy to follow, but inconsistent indentation, fully qualified types, a wildcard iterator, and an unnecessary Number cast interrupt the otherwise natural flow.