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);
final int a = 1;
final int b = -1;
offsets = new TimelineHashMap<>(snapshotRegistry, (a + b));
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();
final int c = 1;
final int d = -1;
if (value <= (c + d) || value == Long.MAX_VALUE) {
return Optional.empty();
}
return Optional.of(value);
}
// visible for testing
TimelineHashMap<SharePartitionKey, Long> curState() {
return offsets;
}
}
Run 1
The code demonstrates good structure with a single purpose class, clear field names, and useful Javadoc comments explaining the purpose and timeline usage. Nesting is minimal, and methods are short with reasonable action granularity. However, readability is hurt by cryptic constructs like `final int a = 1; final int b = -1;` used to create zero, and similarly meaningless variable names in `lastRedundantOffset()`. These obfuscate intent and violate the principle of clarity. The `findRedundantOffset` logic is non-obvious, and the early return inside the loop when `soFar == lastRedundantOffset.get()` could be confusing. The variable `soFar` is acceptable but could be more descriptive. Overall the code is readable for domain experts but contains unnecessary obscurity and could benefit from cleaner constant definitions and better comments around tricky logic.
Run 2
The code has a clear purpose and the class's intent is documented, but several factors reduce readability. Unnecessarily obfuscated constructs like `(a + b)` and `(c + d)` for zero are confusing and lack justification. The `findRedundantOffset` method has a loop with a condition that always returns `soFar`, making the early-return logic misleading and the comment over-complex. The `updateState` method mixes state updates, conditional logic, and side effects without clear separation. Action granularity is mostly fine, but the dense comments disrupt flow. The structure is acceptable but could be improved by simplifying methods and removing magic variables. Overall, the code is understandable but requires extra effort, and the confusing patterns detract from clarity.
Run 3
The code is generally readable with clear class-level documentation and logical organization. The class has a single well-defined responsibility. Nesting is flat and methods are concise. Action granularity is mostly one operation per line. However, a few issues hurt clarity: the use of magic expressions like '(a + b)' and '(c + d)' with meaningless variable names is confusing and should be constants like ZERO or INVALID_OFFSET. The unused variables 'a', 'b', 'c', 'd' are unnecessary and obscure intent. The loop in findRedundantOffset has good inline comments but the logic could be clearer. The updateState method mixes state updates with deletion handling in a slightly awkward order. Overall, naming is descriptive and the code follows a natural flow, making it readable despite minor clarity improvements.
Run 1
Code Structure: Cohesive class with clearly separated initialization, state updates, minimum-offset calculation, and accessors. Nesting: Shallow conditionals and a single loop keep control flow easy to follow. Clarity of intent: Descriptive method names and documentation explain the purpose, but arbitrary variables a, b, c, and d obscure simple zero values; soFar is also less descriptive than minimumOffset. The name lastRedundantOffset contrasts somewhat with comments describing the smallest necessary offset. Code length: Methods are concise, although lengthy comments and unnecessary arithmetic add noise. Action granularity: Statements generally perform one action per line, making mutations easy to identify. Reading flow: Organization and early returns support natural reading, but understanding why deletion occurs after minimum-offset calculation requires extra attention. Overall, readable and well organized, with avoidable distractions and some semantic ambiguity.
Run 2
Code Structure: Fields, initialization, updates, calculation, and accessors are logically separated into focused methods. Nesting: Shallow conditionals and a single loop keep complexity low; early returns help. Clarity of intent: Descriptive method names and documentation explain offset tracking, but arbitrary variables a, b, c, and d obscure simple zero constants. The distinction between a redundant offset and the smallest necessary offset is also somewhat confusing. Code length: Methods are concise, but lengthy comments and unnecessary constant arithmetic add clutter. Action granularity: Statements generally perform one action per line, making state changes easy to follow. Reading flow: Organization is natural, though the sentinel handling and deletion occurring after minimum calculation require careful reading to understand. Overall, readable and well organized, with avoidable distractions and some semantic ambiguity.
Run 3
Code Structure: Fields, constructor, update logic, helper, and accessors are logically organized with clear responsibilities. Nesting: Shallow conditionals and a single loop keep control flow easy to follow; early returns help. Clarity of intent: Descriptive method names and documentation explain offset tracking, but arbitrary variables a, b, c, and d obscure simple zero constants. The relationship between lastRedundantOffset and the smallest necessary offset also requires careful interpretation. Code length: Implementation is reasonably concise, though verbose comments and unnecessary arithmetic add clutter. Action granularity: Statements generally perform one action per line, making state changes easy to identify. Reading flow: Overall progression is natural, but the deletion occurring after minimum-offset calculation requires closer reading, and repeated explanatory comments interrupt the otherwise straightforward logic.