Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java (Java)

Code before:

/**
 * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
 * {@link org.apache.kafka.streams.KafkaStreams} constructor or the {@link KafkaStreamsNamedTopologyWrapper} constructor (deprecated)
 * will determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
 * topology builders via the {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} constructor for DSL applications,
 * or the {@link Topology#Topology(TopologyConfig)} for PAPI applications.
 * <p>
 * Note that some configs, such as the {@code processor.wrapper.class} config, can only take effect while the
 * topology is being built, which means they have to be passed in as a TopologyConfig to the
 * {@link Topology#Topology(TopologyConfig)} constructor (PAPI) or the
 * {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} constructor (DSL).
 * If they are only set in the configs passed in to the KafkaStreams constructor, it will be too late for them
 * to be applied and the config will be ignored.
 */
@SuppressWarnings("deprecation")
public final class TopologyConfig extends AbstractConfig {

    public static class InternalConfig {
        // Cf https://issues.apache.org/jira/browse/KAFKA-19668
        public static final String DISABLE_PROCESS_PROCESSVALUE_FIX = "__disable.process.processValue.fix__";

        public static boolean getBoolean(final Map<String, Object> configs, final String key, final boolean defaultValue) {
            final Object value = configs.getOrDefault(key, defaultValue);
            if (value instanceof Boolean) {
                return (boolean) value;
            } else if (value instanceof String) {
                return Boolean.parseBoolean((String) value);
            } else {
                log.warn(
                    "Invalid value ({}) on internal configuration '{}'. Please specify a true/false value.",
                    value,
                    key
                );
                return defaultValue;
            }
        }
    }

    private static final ConfigDef CONFIG;
    static {
        CONFIG = new ConfigDef()
            .define(PROCESSOR_WRAPPER_CLASS_CONFIG,
                    Type.CLASS,
                    NoOpProcessorWrapper.class.getName(),
                    Importance.LOW,
                    PROCESSOR_WRAPPER_CLASS_DOC)
            .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
                Type.INT,
                null,
                Importance.LOW,

Review comment (suggestion):
Is the cast required, wouldn't the JVM use autoboxing to covert between `Boolean` and `boolean`

Code after:

/**
 * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
 * {@link org.apache.kafka.streams.KafkaStreams} constructor or the {@link KafkaStreamsNamedTopologyWrapper} constructor (deprecated)
 * will determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
 * topology builders via the {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} constructor for DSL applications,
 * or the {@link Topology#Topology(TopologyConfig)} for PAPI applications.
 * <p>
 * Note that some configs, such as the {@code processor.wrapper.class} config, can only take effect while the
 * topology is being built, which means they have to be passed in as a TopologyConfig to the
 * {@link Topology#Topology(TopologyConfig)} constructor (PAPI) or the
 * {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} constructor (DSL).
 * If they are only set in the configs passed in to the KafkaStreams constructor, it will be too late for them
 * to be applied and the config will be ignored.
 */
@SuppressWarnings("deprecation")
public final class TopologyConfig extends AbstractConfig {

    public static class InternalConfig {
        // Cf https://issues.apache.org/jira/browse/KAFKA-19668
        public static final String ENABLE_PROCESS_PROCESSVALUE_FIX = "__enable.process.processValue.fix__";

        public static boolean getBoolean(final Map<String, Object> configs, final String key, final boolean defaultValue) {
            final Object value = configs.getOrDefault(key, defaultValue);
            if (value instanceof Boolean) {
                return (boolean) value;
            } else if (value instanceof String) {
                return Boolean.parseBoolean((String) value);
            } else {
                log.warn(
                    "Invalid value ({}) on internal configuration '{}'. Please specify a true/false value.",
                    value,
                    key
                );
                return defaultValue;
            }
        }
    }

    private static final ConfigDef CONFIG;
    static {
        CONFIG = new ConfigDef()
            .define(PROCESSOR_WRAPPER_CLASS_CONFIG,
                    Type.CLASS,
                    NoOpProcessorWrapper.class.getName(),
                    Importance.LOW,
                    PROCESSOR_WRAPPER_CLASS_DOC)
            .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
                Type.INT,
                null,
                Importance.LOW,