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 defaultAccount) {
Objects.requireNonNull(defaultAccount);
offsets = new TimelineHashMap<>(defaultAccount, 0);
lastRedundantOffset = new TimelineLong(defaultAccount);
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 createCount(SharePartitionKey map, long source, boolean location) {
lastRedundantOffset.set(Math.min(lastRedundantOffset.get(), source));
offsets.put(map, source);
Optional<Long> primarySession = validateRequest();
primarySession.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 (location) {
offsets.remove(map);
}
}
private Optional<Long> validateRequest() {
if (offsets.isEmpty()) {
return Optional.empty();
}
long entry = Long.MAX_VALUE;
for (long window : offsets.values()) {
// Get min offset among latest offsets
// for all share keys in the internal partition.
entry = Math.min(entry, window);
// 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 (entry == lastRedundantOffset.get()) {
return Optional.of(entry);
}
}
return Optional.of(entry);
}
/**
* 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> validateBalance() {
long token = lastRedundantOffset.get();
if (token <= 0 || token == Long.MAX_VALUE) {
return Optional.empty();
}
return Optional.of(token);
}
// visible for testing
TimelineHashMap<SharePartitionKey, Long> saveData() {
return offsets;
}
}
Run 1
The code is difficult to read due to misleading naming and unclear intent. Methods like createCount, validateRequest, and validateBalance have confusing names that don't describe their actual behavior (e.g., createCount updates offsets and computes a minimum; validateRequest does not validate a request). Parameters are also named poorly: 'map' represents a key, 'source' an offset, and 'location' means isDelete. The logic in createCount updates lastRedundantOffset before removing deleted entries, which may lead to incorrect state. Nesting is acceptable and code length is moderate, but unclear naming and ambiguous purpose severely hinder readability. Despite some helpful comments, the code's overall structure and naming make it hard to understand and maintain.
Run 2
The code has severe readability issues. The method and parameter names are misleading: 'createCount' actually updates offsets, 'validateRequest' calculates minimum offsets, and 'validateBalance' retrieves the last redundant offset. Single-letter or meaningless parameter names like 'map', 'source', and 'location' obscure intent. The naming suggests an obfuscation attempt rather than clear purpose. While structure and nesting are acceptable, the mismatch between comments and actual parameter names adds confusion. There are also logical issues: 'createCount' removes offsets after inserting them for delete operations, and 'validateRequest' appears to return early with the minimum only if it equals the last redundant offset, contradicting the stated purpose of finding the global minimum. The code is concise but at the cost of clarity. The reading flow is disrupted by semantic noise. The comments attempt to explain but conflict with variable names. This code requires significant effort to understand actual behavior versus intended behavior, making it nearly unreadable without substantial refactoring.
Run 3
The code has a clear intent but suffers from poor naming and confusing method names. For example, methods like 'createCount', 'validateRequest', 'validateBalance', and 'saveData' do not reflect their actual operations, making the API misleading. Variable names such as 'map', 'source', and 'location' obscure their purposes (key, offset, isDelete). The comments are helpful but are undermined by the poor names. While the code is not deeply nested and generally has reasonable line-level readability, the ambiguous naming heavily impacts clarity. The logic in 'validateRequest' is not as simple as the method name suggests. Overall, the code is understandable with effort but not immediately clear, leading to a moderate score.
Run 1
Code Structure: Fields, initialization, updates, minimum calculation, and accessors are logically separated into small methods. Nesting: Shallow loops and conditionals, with early returns, make control flow easy to follow. Clarity of intent: The main weakness is misleading naming: createCount, validateRequest, validateBalance, saveData, defaultAccount, and location do not describe their offset-tracking roles. Javadoc parameter names disagree with the actual signature, and comments reference nonexistent names such as updateState and soFar. Code length: Executable code is concise, but lengthy explanatory comments outweigh the implementation and introduce inconsistencies. Action granularity: Statements generally perform one action per line, with straightforward expressions. Reading flow: Method organization is natural, but readers must repeatedly translate generic identifiers and reconcile comments with implementation, especially around deletion and redundant-offset semantics.
Run 2
Code Structure: Fields, initialization, updates, calculation, and accessors are logically separated into small methods. Nesting: Mostly flat, with early returns and only one conditional inside a loop. Clarity of intent: Meaningful class and field names help, but misleading method names such as createCount, validateRequest, and validateBalance obscure offset management. Generic parameter and variable names further weaken clarity, while Javadoc parameters and comment references do not match the actual identifiers. Code length: Implementation is compact, but lengthy, occasionally repetitive comments increase reading effort. Action granularity: Statements generally perform one action per line, making state changes easy to follow. Reading flow: Formatting and method ordering are natural, though inconsistent terminology and the need to reconcile comments with code interrupt comprehension.
Run 3
Code Structure: State, initialization, updates, calculation, and accessors are logically separated into small methods. Nesting: Shallow conditionals and a single loop make control flow easy to follow. Clarity of intent: Class and field names convey offset tracking, but misleading method names such as createCount, validateRequest, and validateBalance obscure their purposes. Parameters and locals such as map, source, location, and token further weaken clarity. Javadoc parameter names do not match the signature, and comments reference nonexistent identifiers such as updateState and soFar. Code length: Implementation is reasonably concise, although lengthy comments add reading overhead. Action granularity: Statements generally perform one action per line. Reading flow: Method ordering and early returns support natural reading, but inconsistent terminology and stale documentation force readers to reconcile descriptions with actual behavior.