Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractConfigurableStoreFactory.java (Java)

Code before:
import org.apache.kafka.streams.state.DslStoreSuppliers;

import java.util.HashSet;
import java.util.Set;

public abstract class AbstractConfigurableStoreFactory implements StoreFactory {
    private final Set<String> connectedProcessorNames = new HashSet<>();
    private DslStoreSuppliers dslStoreSuppliers;
    private DslStoreFormat dslStoreFormat;

    public AbstractConfigurableStoreFactory(final DslStoreSuppliers initialStoreSuppliers) {
        this.dslStoreSuppliers = initialStoreSuppliers;
    }

    @Override
    public void configure(final StreamsConfig config) {
        if (dslStoreSuppliers == null) {
            dslStoreSuppliers = config.getConfiguredInstance(
                StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG,
                DslStoreSuppliers.class,
                config.originals()
            );
        }
        if (dslStoreFormat == null) {
            final String dslStoreFormatValue = config.getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
            if (dslStoreFormatValue.equalsIgnoreCase("HEADERS")) {
                dslStoreFormat = DslStoreFormat.HEADERS;
            } else { // DEFAULT
                dslStoreFormat = DslStoreFormat.TIMESTAMPED;
            }
        }

    }

    @Override
    public Set<String> connectedProcessorNames() {
        return connectedProcessorNames;
    }

    public DslStoreFormat dslStoreFormat() {
        return dslStoreFormat;
    }

    protected DslStoreSuppliers dslStoreSuppliers() {
        if (dslStoreSuppliers == null) {
            throw new IllegalStateException("Expected configure() to be called before using dslStoreSuppliers");
        }
        return dslStoreSuppliers;
    }
}


Review comment (suggestion):
```suggestion
            if (dslStoreFormatValue.equalsIgnoreCase(StreamsConfig.DSL_STORE_FORMAT_HEADERS)) {
```

Code after:
import org.apache.kafka.streams.state.DslStoreSuppliers;

import java.util.HashSet;
import java.util.Set;

public abstract class AbstractConfigurableStoreFactory implements StoreFactory {
    private final Set<String> connectedProcessorNames = new HashSet<>();
    private DslStoreSuppliers dslStoreSuppliers;
    private DslStoreFormat dslStoreFormat;

    public AbstractConfigurableStoreFactory(final DslStoreSuppliers initialStoreSuppliers) {
        this.dslStoreSuppliers = initialStoreSuppliers;
    }

    @Override
    public void configure(final StreamsConfig config) {
        if (dslStoreSuppliers == null) {
            dslStoreSuppliers = config.getConfiguredInstance(
                StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG,
                DslStoreSuppliers.class,
                config.originals()
            );
        }
        final String dslStoreFormatValue = config.getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
        if (dslStoreFormatValue.equalsIgnoreCase(StreamsConfig.DSL_STORE_FORMAT_HEADERS)) {
            dslStoreFormat = DslStoreFormat.HEADERS;
        }
        // else dslStoreFormat remains null and the lower layers decide between PLAIN and TIMESTAMPED
    }

    @Override
    public Set<String> connectedProcessorNames() {
        return connectedProcessorNames;
    }

    public DslStoreFormat dslStoreFormat() {
        return dslStoreFormat;
    }

    protected DslStoreSuppliers dslStoreSuppliers() {
        if (dslStoreSuppliers == null) {
            throw new IllegalStateException("Expected configure() to be called before using dslStoreSuppliers");
        }
        return dslStoreSuppliers;
    }
}