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 name, final Map<String, String> topicConfigs, final long retentionMs) {
super(name, topicConfigs);
this.retentionMs = retentionMs;
}
/**
* 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> properties(final Map<String, String> defaultProperties, final long additionalRetentionMs) {
// internal topic config overridden rule: library overrides < global config overrides < per-topic config overrides
final Map<String, String> topicConfig = new HashMap<>(WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES);
topicConfig.putAll(defaultProperties);
topicConfig.putAll(topicConfigs);
if (!topicConfigs.containsKey(TopicConfig.RETENTION_MS_CONFIG)) {
long retentionValue;
try {
retentionValue = Math.addExact(retentionMs, additionalRetentionMs);
} catch (final ArithmeticException swallow) {
retentionValue = Long.MAX_VALUE;
}
topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionValue));
}
return topicConfig;
}
@Override
public boolean equals(final Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
final WindowedChangelogTopicConfig that = (WindowedChangelogTopicConfig) o;
return Objects.equals(name, that.name) &&
Objects.equals(topicConfigs, that.topicConfigs) &&
Objects.equals(retentionMs, that.retentionMs) &&
Objects.equals(enforceNumberOfPartitions, that.enforceNumberOfPartitions);
}
@Override
public int hashCode() {
return Objects.hash(name, topicConfigs, retentionMs, enforceNumberOfPartitions);
}
@Override
public String toString() {
return "WindowedChangelogTopicConfig(" +
"name=" + name +
", topicConfigs=" + topicConfigs +
", retentionMs=" + retentionMs +
", enforceNumberOfPartitions=" + enforceNumberOfPartitions +
")";
}
}
Run 1
The code is well-structured and readable. Class hierarchy and static initialization are clear, with logical organization. Nesting is minimal, with only one try-catch block in the properties method. The intent is clearly documented with Javadoc and inline comments explaining the override precedence rules. Code length is appropriate for the functionality. Action granularity is mostly good, with one operation per line. Reading flow is natural, with methods in a logical order. Minor deductions: the static initializer could be simplified with a method, and the try-catch variable naming ('swallow') is slightly unconventional but acceptable. Overall, this is highly maintainable code.
Run 2
The code is well-structured and readable. It follows a clear logical organization: constants, fields, constructor, then overridden methods. Nesting is minimal, with only one level inside the properties method. Intent is mostly clear, especially the static initializer for default overrides and the documented retention logic. Action granularity is appropriate, with one operation per line in most places. The reading flow is natural, though the long putAll chains in properties() could be slightly more explicit. The main weakness is the try-catch block with a swallowed ArithmeticException — the variable name 'swallow' clarifies intent, but the logic remains slightly dense. Field visibility and missing final modifiers on class/fields slightly reduce clarity. Overall, the code is concise, well-documented, and maintainable with only minor readability improvements possible.
Run 3
The code is highly readable with clear structure and intent. It has a logical organization, with constants, fields, constructor, and overridden methods well-separated. Nesting is minimal, with early returns in equals() and only one level of conditional logic in properties(). The purpose is unambiguous, supported by meaningful Javadoc and variable names. Code length is appropriate—no excessive verbosity or duplication. Action granularity is mostly one action per line, though the long put() line in the static block and chained equals() conditions slightly reduce readability. Reading flow is natural, guiding the reader from setup through configuration to standard object methods. Minor improvements could include breaking the long static initialization line and extracting the overflow handling into a helper method.
Run 1
Code Structure: Well-organized class with defaults, constructor, configuration logic, and standard object methods clearly separated. Nesting: Shallow overall; only the retention check contains a nested try/catch. Clarity of intent: Descriptive names and documentation explain configuration precedence and retention adjustment, although inherited fields require external context and the exception variable 'swallow' is unconventional. Code length: Appropriately concise, with modest boilerplate for equals, hashCode, and toString. Action granularity: Statements generally perform one action per line; long configuration expressions slightly reduce scanability. Reading flow: Natural progression from initialization to behavior and utility methods, aided by whitespace and early returns. Minor drawbacks include several long lines, limited import spacing, and unnecessarily boxed primitive comparisons in equals.
Run 2
Code Structure: Well-organized class with distinct initialization, configuration, and standard object methods. Nesting: Mostly flat, with only a shallow if and try/catch in properties. Clarity of intent: Descriptive names and comments explain configuration precedence, retention adjustment, and overflow handling; inherited fields require some external context. Code length: Appropriately concise, though standard object-method boilerplate adds length. Action granularity: Each statement performs a clear action, with no compressed multi-statement lines. Reading flow: Natural progression from defaults and constructor to behavior and object methods. Minor readability issues include long declarations and configuration lines, slightly awkward documentation, and using Objects.equals for primitive retention values rather than direct comparison.
Run 3
Code Structure: Well-organized class with defaults, state, construction, configuration logic, and standard object methods clearly separated. Nesting: Shallow overall; only the retention calculation introduces a try/catch inside an if. Clarity of intent: Descriptive names and documentation explain configuration precedence and retention behavior, although the overflow fallback could use an explanatory comment. Code length: Appropriately concise, with conventional equals, hashCode, and toString boilerplate. Action granularity: Each statement performs one clear action; long expressions slightly reduce scanability. Reading flow: Initialization and configuration merging follow a natural sequence, and equals uses clear early returns. Minor readability drawbacks include long lines, sparse spacing around imports, and the vague exception variable name 'swallow'.