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 generally readable and well-organized. The class has a clear single purpose, with a logical static initializer for default overrides and a focused constructor. The `properties` method follows a clear precedence chain (defaults -> global -> per-topic) and includes an explanatory comment. Nesting is minimal, with only one level in the `equals` method and a try-catch. Naming is descriptive, and action granularity is good — most lines perform one logical step. However, the long static initializer in the class body slightly interrupts reading flow, and the `retentionMs` property lacks an explanatory comment at its declaration, which would clarify its role. The equals/hashCode/toString methods are standard but verbose, adding length without adding much readability benefit. Overall, the code is maintainable and straightforward for a Kafka internals developer.
Run 2
The code is well-structured, with clear separation of concerns and consistent formatting. Class purpose is documented, static initialization is clean, and the implementation follows a logical flow. Nesting is minimal, with only one level inside the properties() method. Variable names are descriptive, and the equals/hashCode/toString implementations are standard and readable. The comment about configuration override precedence is helpful. Minor deductions: multiple map operations on adjacent lines could be slightly more separated, and the `properties()` method performs both map merging and conditional retention logic, making it moderately dense. Overall, the code is highly readable with minimal ambiguity.
Run 3
The code is generally well-structured and readable. The class has a clear single responsibility, and methods are concise with flat control flow. The static initializer for default overrides is logically separated, and the properties() method is straightforward: it builds a configuration map in three clear steps with a helpful comment explaining precedence. Nesting is minimal, and actions are granular—one major operation per line. The only improvement would be slightly better inline documentation around the override rule and the treatment of retentionMs defaulting, but the existing code remains easy to follow and maintain.
Run 1
Code Structure: Well organized into default initialization, construction, configuration generation, and standard object methods. Nesting: Shallow overall; only the retention condition contains a nested try/catch. Clarity of intent: Descriptive names, Javadoc, and an explicit configuration-precedence comment make the purpose clear, though the overflow fallback could use a brief explanation. Code length: Appropriately concise, with some unavoidable boilerplate in equals, hashCode, and toString. Action granularity: Each statement performs a distinct operation, making configuration assembly easy to follow. Reading flow: Defaults, overrides, retention adjustment, and return follow a natural sequence. Minor readability issues include long declarations and expressions, slightly awkward comment wording, and reliance on inherited fields whose definitions are not visible.
Run 2
Code Structure: Well organized into default initialization, construction, property resolution, and standard object methods. Nesting: Shallow overall; the conditional containing try/catch is manageable. Clarity of intent: Descriptive identifiers and documentation explain configuration precedence and retention adjustment, although inherited fields require superclass context and the overflow fallback could use a brief explanation. Code length: Appropriately concise, with conventional equals, hashCode, and toString boilerplate. Action granularity: Each statement performs one clear operation; configuration merging is easy to follow. Reading flow: Methods progress naturally, and early returns simplify equals. Minor readability drawbacks include long declarations and configuration expressions, inconsistent spacing between imports and package declaration, and slightly awkward comment wording.
Run 3
Code Structure: Well-organized class with defaults, constructor, configuration logic, and standard object methods clearly separated. Nesting: Shallow overall; the nested try/catch inside the retention condition remains easy to follow. Clarity of intent: Descriptive names and documentation explain configuration precedence and retention calculation, though the exception variable 'swallow' is less informative and overflow handling could use a brief comment. Code length: Appropriately concise, with some conventional boilerplate and long declarations. Action granularity: Statements generally perform one action per line; chained equality checks and string concatenation are formatted clearly. Reading flow: Initialization leads naturally into behavior and utility methods. Minor drawbacks include tightly packed imports, long lines, and unnecessary Objects.equals usage for primitive retentionMs.