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 and readable, with clear separation of concerns and a logical progression from static initialization through constructor to methods. Nesting is minimal, limited to a single conditional and try-catch block. Intent is explicit, especially in the `properties` method, which has a descriptive comment explaining the compaction/deletion retention calculation. The code is concise and avoids unnecessary verbosity, though a few minor improvements could be made, such as the slightly awkward parameter name `swallow` in the empty catch block. Overall, the class maintains a clean, flat structure with good method granularity and a natural reading flow that follows Java conventions effectively.
Run 3
The code is well-structured and readable. The class has a clear single responsibility, and the static initializer cleanly organizes default overrides without cluttering the constructor. Nesting is minimal: the main method only has one conditional block and a try-catch, both shallow. Intent is clearly conveyed through descriptive names and a useful Javadoc comment explaining the retention logic. Code length is appropriate and avoids unnecessary verbosity. Action granularity is generally one logical step per line, improving scannability. Reading flow is natural: static setup, constructor, overridden methods, then equals/hashCode/toString. One minor issue is the variable name 'swallow' in the catch block, which is somewhat unusual, but the code otherwise follows consistent conventions and is easy to follow.
Run 1
Code Structure: Logically organized into default initialization, construction, configuration generation, and standard object methods. Nesting: Shallow overall; the conditional containing try/catch is straightforward. Clarity of intent: Descriptive names, class documentation, and the configuration-precedence comment explain the purpose well. Overflow handling is explicit, although the exception name 'swallow' is less informative. Code length: Appropriately concise, with some routine verbosity in equals, hashCode, and toString. Action granularity: Statements generally perform one action per line; long configuration expressions slightly reduce scanability. Reading flow: Natural progression and early returns make the code easy to follow. Minor drawbacks include lengthy declarations and expressions, inherited fields requiring superclass context, and unnecessary Objects.equals calls for primitive retention values.
Run 2
Code Structure: Well organized into defaults initialization, construction, configuration generation, and standard object methods. Nesting: Shallow overall; the nested try/catch within the retention check is straightforward. Clarity of intent: Descriptive names, Javadoc, and the explicit configuration-precedence comment explain the purpose well, though the overflow fallback could benefit from a brief rationale. Code length: Appropriately concise, with conventional equals, hashCode, and toString boilerplate; several long declarations and expressions slightly hinder scanning. Action granularity: Statements generally perform one action per line, and configuration merging is clearly separated. Reading flow: Natural progression and early returns make the code easy to follow. Minor distractions include inconsistent import grouping, excessive blank lines between related putAll calls, and Objects.equals for primitive retention values.
Run 3
Code Structure: Logically organized into defaults, construction, configuration generation, and standard object methods. Nesting: Shallow overall; only the retention condition contains a nested try/catch. Clarity of intent: Descriptive names and documentation explain configuration precedence and retention handling, though the overflow fallback could benefit from a brief comment. Code length: Appropriately concise, with conventional equals, hashCode, and toString boilerplate. Action granularity: Each statement performs a distinct action; long expressions remain understandable but would benefit from wrapping. Reading flow: Natural progression and early returns make the class easy to follow. Minor readability drawbacks include lengthy signatures and configuration lines, slightly awkward documentation wording, and Objects.equals for primitive retention values instead of direct comparison.