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 snapshot) {
Objects.requireNonNull(snapshot);
offsets = new TimelineHashMap<>(snapshot, 0);
lastRedundantOffset = new TimelineLong(snapshot);
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 update(SharePartitionKey key, long off, boolean is) {
lastRedundantOffset.set(Math.min(lastRedundantOffset.get(), off));
offsets.put(key, off);
Optional<Long> redundant = find();
redundant.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 (is) {
offsets.remove(key);
}
}
private Optional<Long> find() {
if (offsets.isEmpty()) {
return Optional.empty();
}
long so = Long.MAX_VALUE;
for (long off2 : offsets.values()) {
// Get min offset among latest offsets
// for all share keys in the internal partition.
so = Math.min(so, off2);
// 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 (so == lastRedundantOffset.get()) {
return Optional.of(so);
}
}
return Optional.of(so);
}
/**
* 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> last() {
long val = lastRedundantOffset.get();
if (val <= 0 || val == Long.MAX_VALUE) {
return Optional.empty();
}
return Optional.of(val);
}
// visible for testing
TimelineHashMap<SharePartitionKey, Long> cur() {
return offsets;
}
}
Run 1
The code suffers from poor naming and unclear intent, significantly hurting readability. Variables like 'off', 'is', 'so', 'off2', and 'cur' are cryptic and give no indication of their purpose, violating the 'clarity of intent' criterion. This forces the reader to constantly refer back to comments or method signatures. The 'find' method contains convoluted logic: the early return condition inside the loop makes the loop's purpose and the method's final result confusing. The comment explains a 'contiguous prefix' concept, but the code logic does not clearly mirror that explanation, making the relationship ambiguous. Furthermore, the 'update' method performs multiple actions (setting the minimum, putting the offset, calling find, conditionally removing) without clear transitional logic, and the sequence feels disorganized. While the file has logical package/import structure and basic Javadocs, the poor naming, ambiguous conditionals, and multi-purpose methods greatly reduce readability.
Run 2
The code has a clear class purpose and some helpful comments, but readability suffers from several issues. Naming is inconsistent and vague: parameters like 'off', 'is', 'so', 'off2', and method 'find'/'cur' obscure intent. The update method performs multiple actions (setting offsets, calculating redundancy, handling deletion) in a non-logical order, and the deletion check occurs after offset insertion, which can be confusing. The find method contains misleading logic: it iterates over all values but returns early if 'so' equals lastRedundantOffset, making the loop partially redundant. Comments are verbose yet do not clarify the algorithm's core logic. The method 'last' intermixes validity checks with retrieval. Overall, the code is functional but requires careful reading to understand the offset-tracking algorithm, indicating poor action granularity and clarity of intent.
Run 3
The code has a clear class-level purpose and reasonable structure, but its readability is undermined by several issues. Conciseness is a point of concern — method and parameter names like 'update', 'off', 'is', 'find', and 'cur' are too abbreviated, creating ambiguity about intent. The comments are sometimes helpful in explaining behavior but also include an odd inline example that adds visual noise. The 'find' method has confusing naming (using 'so' and 'off2') and is not easy to follow. The sequence of operations in 'update' is somewhat non-intuitive — setting offsets before checking and removing deleted keys makes the logic harder to grasp. Nesting is not deep, but action granularity is mixed, with some lines doing multiple actions. Overall, the code is understandable if the reader works through it, but it does not flow naturally and would benefit from clearer naming and cleaner method decomposition.
Run 1
Code Structure: Well organized into initialization, updates, minimum-offset calculation, queries, and test access. Nesting: Shallow, with early returns and only a loop containing one conditional. Clarity of intent: Class documentation and descriptive field names explain the overall purpose, but names such as off, is, so, off2, find, last, and cur obscure local intent. Javadoc parameter names do not match the update signature, and comments reference updateState and soFar rather than actual identifiers. Code length: Implementation is reasonably concise, though lengthy comments repeat concepts and interrupt scanning. Action granularity: Generally one action per line, with straightforward assignments and calls. Reading flow: Method sequencing and control flow are natural, but calculating the minimum before removing a deleted key requires careful interpretation, and inconsistent terminology around redundant versus necessary offsets adds ambiguity.
Run 2
Code Structure: Fields, initialization, updates, minimum calculation, and accessors are logically separated into small methods. Nesting: Shallow control flow, with only an if inside a loop, makes navigation easy. Clarity of intent: Class documentation and descriptive field names explain the overall purpose, but names such as find, last, cur, off, is, so, and off2 obscure specific roles. Documentation references parameter names and identifiers that do not match the implementation, further reducing clarity. Code length: Executable code is concise, although lengthy, sometimes repetitive comments interrupt readability. Action granularity: Statements generally perform one action per line, with no compressed multi-statement lines. Reading flow: The method order and early returns support natural reading, but understanding why the minimum is calculated before deletion requires extra attention, and terminology around redundant versus necessary offsets is somewhat confusing.
Run 3
Code Structure: Cohesive class with clearly separated initialization, updates, minimum-offset calculation, and accessors. Nesting: Shallow conditionals and a single loop make control flow easy to follow. Clarity of intent: Descriptive field names and explanatory comments help, but names such as find, last, cur, off, is, so, and off2 obscure purpose. Javadoc parameter names do not match the actual signature, and comments reference nonexistent names such as updateState and soFar. Code length: Methods are concise, although some comments are lengthy or repetitive. Action granularity: Each statement generally performs one action, with no crowded multi-statement lines. Reading flow: The overall sequence is natural, but inconsistent terminology around redundant versus necessary offsets and the deletion occurring after minimum calculation require extra thought.