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. It has a clear class-level constant initialization, a sensible constructor, and a focused override method for properties. Nesting is minimal, with only a single if-statement and a try-catch. Intent is clear: the class configures windowed changelog topics with retention logic. The properties method is concise and follows a logical override sequence (defaults → global → per-topic). Action granularity is generally one operation per line. Reading flow is natural, with comments explaining non-obvious behavior like retention calculation and the override hierarchy. Minor deductions: the static initialization block slightly interrupts flow, and the equality/toString methods add length but are standard boilerplate. Overall, it is highly readable.
Run 3
The code is generally readable with clear separation of concerns, a logical static initialization block, and straightforward method implementations. The `properties` method effectively communicates the precedence rules via comments and sequential `putAll` calls. However, some minor readability issues exist: the method name `properties` doesn't clearly convey it returns a full merged config, the variable `swallow` is ambiguous (though commonly used for intentionally ignored exceptions), and the long line in the static block exceeds typical readability width. The class structure is flat, with low nesting and clear intent, but some lines could be broken for better readability. Overall, it's well-structured and maintainable, though not perfectly polished.
Run 1
Code Structure: Well-organized class with defaults initialization, constructor, configuration logic, and standard object methods clearly separated. Nesting: Mostly flat; only the retention calculation nests a try/catch inside an if, which remains easy to follow. Clarity of intent: Descriptive names, Javadoc, and the configuration-precedence comment explain the purpose and behavior. Overflow handling is explicit, though the catch variable name 'swallow' is less informative. Code length: Appropriately concise overall, with some conventional boilerplate and lengthy declarations. Action granularity: Statements generally perform one action per line; chained string concatenations and equality checks remain readable. Reading flow: Configuration assembly follows a natural sequence from defaults to overrides to retention adjustment. Minor drawbacks include long lines, sparse spacing around imports, and reliance on inherited fields and constants for context.
Run 2
Code Structure: Logically organized into default initialization, construction, property assembly, and standard object methods. Nesting: Shallow overall; the try/catch inside a single conditional remains easy to follow. Clarity of intent: Descriptive identifiers 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 conventional equals, hashCode, and toString boilerplate. Action granularity: Statements generally perform one action per line; a few long expressions reduce scanability slightly. Reading flow: Natural progression and clear spacing make the class easy to navigate, although long signatures and the cleanup-policy assignment would benefit from wrapping.
Run 3
Code Structure: Logically organized into defaults, construction, property configuration, and standard object methods. Nesting: Shallow overall; the retention conditional contains a straightforward try/catch. Clarity of intent: Descriptive names and documentation explain configuration precedence and retention adjustment, though the exception name 'swallow' does not clarify overflow handling. Code length: Appropriately concise, with some verbose constants and routine boilerplate. Action granularity: Each statement performs a distinct action, making state changes easy to follow. Reading flow: Natural progression and useful spacing support readability; a few long declarations and expressions would benefit from wrapping. Overall, the code is highly readable with minor presentation and naming issues.