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;
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 Javadoc and reasonable method separation, but has several readability issues. The updateState method performs multiple actions: setting offsets, calculating redundant offset, conditionally removing keys, and handling deletes—violating single-action granularity. The findRedundantOffset method contains confusing logic with a lengthy inline comment explaining a non-obvious early return condition, making intent unclear. The lastRedundantOffset initialization to Long.MAX_VALUE is a hidden sentinel value requiring comments to understand. The package declaration and imports could benefit from blank lines for separation. Overall nesting is shallow and the reading flow is acceptable, but the redundant offset calculation logic is unnecessarily complex and would benefit from refactoring into smaller, better-named methods with clearer explanation of the edge case handling.
Run 2
The code is well-structured and readable. It has a clear class purpose with a concise Javadoc, and methods are logically organized. Nesting is minimal, with only one level in the loop. Intent is clear throughout, aided by descriptive method names and useful inline comments explaining non-obvious logic. The code is concise without excessive verbosity, and each line generally performs a single action. However, a few points slightly reduce readability: the `findRedundantOffset` method mixes state mutation with its calculation, and some comments are longer than necessary. Overall, the reading flow is natural, and the code is easy to follow.
Run 3
The code is generally well-structured with clear separation of concerns, a single public mutator, and focused private helpers. Javadoc comments explain the purpose and non-obvious behavior. Nesting is minimal, with no deep conditional blocks. Each line performs one logical action. However, readability is slightly reduced by the ambiguous logic in findRedundantOffset: the early return on equality with lastRedundantOffset is not intuitive and relies heavily on comments. Additionally, the field name offsets is vague—it stores the latest offset per key, not a collection of offsets. The initial set of lastRedundantOffset to Long.MAX_VALUE as a sentinel is explained but still a subtle pattern. Overall, the code is readable but would benefit from clearer naming and a more explicit algorithm description.
Run 1
Code Structure: Fields, initialization, state updates, minimum calculation, and accessors are logically separated into focused methods. Nesting: Shallow loops and conditionals, with early returns, keep complexity low. Clarity of intent: Descriptive types, method names, and comments explain offset tracking and snapshot rollback; however, lastRedundantOffset is described as the smallest necessary offset, creating boundary ambiguity, and soFar is vague. Code length: Implementation is compact, but lengthy and occasionally repetitive comments interrupt readability. Action granularity: Statements generally perform one action per line, with straightforward assignments and collection operations. Reading flow: The class progresses naturally from setup to updates and queries. However, calculating the redundant offset before removing a deleted key requires careful interpretation, and its ordering could be explained more explicitly. Overall, the code is readable and well organized, with minor terminology and documentation issues.
Run 2
Code Structure: A focused class with well-separated initialization, state updates, minimum-offset calculation, and accessors. Nesting: Shallow control flow and early returns make methods easy to follow. Clarity of intent: Descriptive names and documentation explain snapshot handling and deletion behavior, but lastRedundantOffset ambiguously denotes a boundary versus an actually redundant offset, and soFar is generic. Code length: Executable code is concise; several comments are repetitive or disproportionately long. Action granularity: Statements generally perform one action per line, with straightforward expressions. Reading flow: Methods follow a natural progression and formatting is consistent, though the subtle ordering of minimum calculation before deletion requires careful reading. The lengthy inline example interrupts the otherwise smooth calculation loop.
Run 3
Code Structure: Fields, constructor, state updates, calculation helper, and accessors are logically organized with clear separation of responsibilities. Nesting: Shallow conditionals and a single loop make control flow easy to follow; early returns avoid unnecessary nesting. Clarity of intent: Descriptive method names and documentation explain snapshot rollback and offset tracking, though lastRedundantOffset versus smallest necessary offset introduces terminology ambiguity, and deletion after calculation requires careful reading. Code length: Implementation is concise, but several comments and accessor documentation are verbose or repetitive. Action granularity: Statements generally perform one action per line; composed Math.min calls remain straightforward. Reading flow: State updates lead naturally into minimum-offset calculation and querying. However, tightly grouped imports, the vague accumulator name soFar, and a lengthy comment inside the loop slightly interrupt readability.