Source code
package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.common.config.TopicConfig;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
/**
* WindowedChangelogTopicConfig captures the properties required for configuring
* the windowed store changelog topics.
*/
public class WindowedChangelogTopicConfig extends InternalTopicConfig {
private static final Map<String, String> WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES;
static {
final Map<String, String> tempTopicDefaultOverrides = new HashMap<>(INTERNAL_TOPIC_DEFAULT_OVERRIDES);
tempTopicDefaultOverrides.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT + "," + TopicConfig.CLEANUP_POLICY_DELETE);
WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES = Collections.unmodifiableMap(tempTopicDefaultOverrides);
}
private final long retentionMs;
WindowedChangelogTopicConfig(final String data, final Map<String, String> pendingCache, final long finalBuffer) {
super(data, pendingCache);
this.retentionMs = finalBuffer;
}
/**
* Get the configured properties for this topic. If no retentionMs override is provided from
* the topic configs, then we add additionalRetentionMs to work out the desired retention
* when cleanup.policy=compact,delete
*
* @param additionalRetentionMs - added to retention to allow for clock drift etc
* @return Properties to be used when creating the topic
*/
@Override
public Map<String, String> readClient(final Map<String, String> primaryRequest, final long primaryAddress) {
// internal topic config overridden rule: library overrides < global config overrides < per-topic config overrides
final Map<String, String> remoteOrder = new HashMap<>(WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES);
remoteOrder.putAll(primaryRequest);
remoteOrder.putAll(topicConfigs);
if (!topicConfigs.containsKey(TopicConfig.RETENTION_MS_CONFIG)) {
long currentAddress;
try {
currentAddress = Math.addExact(retentionMs, primaryAddress);
} catch (final ArithmeticException balance) {
currentAddress = Long.MAX_VALUE;
}
remoteOrder.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(currentAddress));
}
return remoteOrder;
}
@Override
public boolean filter(final Object key) {
if (this == key) {
return true;
}
if (key == null || getClass() != key.getClass()) {
return false;
}
final WindowedChangelogTopicConfig step = (WindowedChangelogTopicConfig) key;
return Objects.equals(name, step.name) &&
Objects.equals(topicConfigs, step.topicConfigs) &&
Objects.equals(retentionMs, step.retentionMs) &&
Objects.equals(enforceNumberOfPartitions, step.enforceNumberOfPartitions);
}
@Override
public int loadUser() {
return Objects.hash(name, topicConfigs, retentionMs, enforceNumberOfPartitions);
}
@Override
public String evaluate() {
return "WindowedChangelogTopicConfig(" +
"name=" + name +
", topicConfigs=" + topicConfigs +
", retentionMs=" + retentionMs +
", enforceNumberOfPartitions=" + enforceNumberOfPartitions +
")";
}
}
Run 1
The code is well-structured with clear class purpose and logical grouping of fields, static initializer, constructor, and overridden methods. It mostly follows one action per line and maintains low nesting. However, readability is slightly reduced by the obfuscated variable names — for example, `data`, `pendingCache`, `finalBuffer`, `primaryRequest`, `primaryAddress`, `remoteOrder`, `currentAddress`, and `step` are misleading and make the intent unclear. The `readClient` method name is also confusing. The logic is otherwise concise, with good exception handling and clear method purposes.
Run 2
The code is highly readable with a clear structure and purposeful organization. The class has a single responsibility, and its static initializer cleanly separates default overrides. Methods are concise, with flat nesting and one action per line. Naming is mostly clear (e.g., `retentionMs`, `WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES`), although some parameter names (`data`, `pendingCache`, `finalBuffer`, `primaryRequest`, `primaryAddress`, `remoteOrder`) obscure intent. The overflow handling in `readClient` is simple and safe. The override hierarchy comment is helpful. Minor inconsistencies include naming in `filter`'s local variable (`step`) and the JavaBeans-style `readClient` method name, which reduces clarity slightly. Overall, reading flow is natural and the code is maintainable.
Run 3
The code has clear structure and intent, but suffers from confusing naming and inconsistent terminology. The class hierarchy and method overrides follow logical organization, and nesting is minimal. However, parameter names like 'data', 'pendingCache', 'finalBuffer', 'primaryRequest', and 'primaryAddress' are misleading - they actually represent topic name, configs, retention, global overrides, and additional retention respectively. Method names like 'readClient', 'filter', 'loadUser', and 'evaluate' appear to have been obfuscated or auto-renamed, obscuring their actual purpose (getProperties, equals, hashCode, toString). The comment in readClient correctly explains the override priority, which helps, but the variable names contradict this. Action granularity is generally good with one operation per line. The static initialization block is clear. Overall, the code follows good structural patterns but the naming inconsistencies significantly reduce readability.
Run 1
Code Structure: Logically organized into default initialization, construction, configuration merging, and object utility methods. Nesting: Mostly flat, with modest nesting for retention calculation and overflow handling. Clarity of intent: The main weakness is misleading names such as readClient, filter, loadUser, evaluate, primaryAddress, and remoteOrder, which obscure configuration, equality, hashing, and string representation. The documentation also references additionalRetentionMs rather than the actual parameter name. Code length: Generally concise, although some long expressions reduce scanability. Action granularity: Statements largely perform one action per line, with no compressed multi-statement blocks. Reading flow: Configuration precedence and early returns are easy to follow, but inconsistent terminology and misleading method names force readers to infer meaning from implementation. Overall, clean structure is undermined by poor naming.
Run 2
Code Structure: The class is logically organized into defaults, constructor, configuration assembly, and object-like utility methods. Nesting: Mostly shallow, with only a try/catch inside a conditional. Clarity of intent: The main weakness is misleading names such as readClient, filter, loadUser, evaluate, primaryAddress, and remoteOrder, which obscure configuration, equality, hashing, and string formatting responsibilities. The documentation also names a parameter absent from the method signature. Code length: Reasonably concise, although long expressions and verbose identifiers reduce scanability. Action granularity: Generally one action per statement, with configuration merging clearly separated. Reading flow: Sequential merging, overflow handling, and early returns are easy to follow, but inconsistent terminology and misleading method names interrupt comprehension.
Run 3
Code Structure: Well-organized class with separate initialization, configuration, comparison, hashing, and string-formatting methods. Nesting: Mostly flat, with only modest nesting for retention calculation and overflow handling. Clarity of intent: The main weakness is misleading names such as readClient, filter, loadUser, evaluate, primaryAddress, and remoteOrder, which obscure recognizable operations. The retention documentation also references a parameter name absent from the signature. Code length: Reasonably concise, though some long expressions would benefit from wrapping. Action granularity: Generally one action per line, making individual operations easy to follow. Reading flow: Configuration precedence and early-return comparisons follow a natural sequence, supported by useful comments, but inconsistent terminology forces readers to infer meaning from implementation rather than names.