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);
(findRedundantOffset()).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;
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 well-structured with clear class responsibilities and good documentation. The package organization and imports are clean. The nesting depth is minimal (one level in findRedundantOffset), and each method performs a single logical action. The intent is clearly communicated through meaningful names like lastRedundantOffset and updateState. However, there are minor readability issues: an unnecessary blank line in updateState, the comment structure in findRedundantOffset could be more concise, and the example in comments slightly interrupts flow. The code is reasonably concise but the lengthy comments in findRedundantOffset could be trimmed. Overall, it's highly readable with clear purpose and good action granularity.
Run 2
The code is generally well-structured with clear intent, good documentation, and logical organization. It uses flat control flow and avoids deep nesting. However, there are several readability issues: the findRedundantOffset() method contains an ineffective early-return condition (soFar == lastRedundantOffset) that doesn't actually skip remaining iterations, and the logic comments are somewhat verbose. The updateState() method performs multiple actions per line (set, put, conditional removal), which slightly reduces action granularity. Variable naming is clear, and the class has a single responsibility. The redundant offset calculation logic could be simplified for better clarity. Overall, the code is maintainable and understandable but could benefit from minor refactoring to improve action separation and method clarity.
Run 3
The code is moderately readable but has several issues. The class has a clear purpose and good documentation comments explaining the timeline objects and redundant offset logic. However, the findRedundantOffset() method contains confusing logic with an early return inside a loop that appears to defeat the purpose of iterating through all offsets. The comment block inside the loop is lengthy and disrupts flow, and the logic comparing soFar to lastRedundantOffset is unclear. There are also minor issues like the stray blank line before findRedundantOffset() call, and the method name curState() is vague. The updateState method mixes multiple concerns (setting offset, calculating minimum, handling delete) which reduces clarity. While the code is relatively flat and each line performs a single action, the intent of the redundant offset calculation is hard to follow without deeper analysis.
Run 1
Code Structure: Well-organized class with focused methods and clearly separated state, updates, calculations, and accessors. Nesting: Shallow control flow; the loop contains only one conditional, and early returns simplify reading. Clarity of intent: Descriptive names and documentation explain offset tracking and snapshot behavior, though 'lastRedundantOffset' versus 'smallest necessary offset' creates some conceptual ambiguity, and 'soFar' is vague. Code length: Implementation is concise, but several comments are verbose or repeat the code. Action granularity: Statements generally perform one action per line; the minimum-update expression is compact and understandable. Reading flow: Mostly natural, with initialization followed by updates and queries. Extra blank lines, unnecessary parentheses around findRedundantOffset(), and calculating redundancy before removing deleted keys introduce minor reading friction.
Run 2
Code Structure: Well-organized class with focused methods and clearly separated state, updates, calculations, and queries. Nesting: Shallow conditionals and a single loop keep control flow easy to follow. Clarity of intent: Descriptive names and documentation explain offset tracking and snapshot behavior, though lastRedundantOffset versus smallest necessary offset introduces boundary ambiguity, and soFar is generic. Code length: Implementation is concise, but lengthy comments repeat concepts and interrupt scanning. Action granularity: Each statement generally performs one action; the Math.min updates and method reference remain understandable. Reading flow: Methods follow a natural progression, with early returns simplifying traversal. Minor distractions include inconsistent whitespace, unnecessary parentheses around findRedundantOffset(), and the need to reason carefully about calculating the minimum before removing a deleted key.
Run 3
Code Structure: Fields, initialization, updates, calculation, and accessors are logically separated into focused methods. Nesting: Shallow conditionals and a single loop keep control flow easy to follow. Clarity of intent: Descriptive method names and documentation explain offset tracking and snapshot behavior, though lastRedundantOffset describes a boundary rather than an obviously redundant record, and soFar is vague. Code length: Implementation is compact, but lengthy comments repeat concepts and interrupt scanning. Action granularity: Each statement generally performs one clear action. Reading flow: Organization and early returns support natural reading, while redundant parentheses, extra blank space, and computing the minimum before removing a deleted key require additional interpretation.