Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java (Java)

Code before:
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized = new MaterializedInternal<>(
            Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("store").withLoggingDisabled(), nameProvider, STORE_PREFIX
        );

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final StateStore wrapped = ((WrappedStateStore) store).wrapped();
        assertThat(wrapped, not(instanceOf(ChangeLoggingKeyValueBytesStore.class)));
    }

    @Test
    public void shouldNotBuildHeadersAwareStoreWithCachingEvenIfExplicitlySet() {
        doReturn("headers")
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized =
            new MaterializedInternal<>(Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("store").withCachingEnabled(), nameProvider, STORE_PREFIX);

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped();
        assertThat(store, instanceOf(MeteredTimestampedKeyValueStoreWithHeaders.class));
        assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.class));
        // Verify that caching layer was NOT added
        assertThat(logging, not(instanceOf(CachingKeyValueStore.class)));
    }

    @Test
    public void shouldCreateHeadersAwareStoreWithProvidedSupplierAndLoggingEnabled() {
        mockKeyValueStoreSupplier();
        doReturn("headers")
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized =
            new MaterializedInternal<>(Materialized.as(keyValueStoreSupplier), nameProvider, STORE_PREFIX);

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped();
        assertThat(innerKeyValueStore.name(), equalTo(store.name()));
        assertThat(store, instanceOf(MeteredTimestampedKeyValueStoreWithHeaders.class));
        assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.class));
    }

    @SuppressWarnings("unchecked")
    private TimestampedKeyValueStore<String, String> getTimestampedStore(
        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized) {
        final KeyValueStoreMaterializer<String, String> materializer = new KeyValueStoreMaterializer<>(materialized);
        materializer.configure(streamsConfig);

Review comment (bug):
It seems that the test will fail because that we only check cache condition for versioned store ?
https://github.com/aliehsaeedii/kafka/blob/560f78cd966899d21ad94501373bc421204898eb/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java#L84-L90

Code after:
        doReturn("headers")
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized = new MaterializedInternal<>(
            Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("store").withLoggingDisabled(), nameProvider, STORE_PREFIX
        );

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final StateStore wrapped = ((WrappedStateStore) store).wrapped();
        assertThat(wrapped, not(instanceOf(ChangeLoggingKeyValueBytesStore.class)));
    }

    @Test
    public void shouldBuildHeadersAwareStoreWithCachingEnabledByDefault() {
        doReturn("headers")
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized =
            new MaterializedInternal<>(Materialized.as("store"), nameProvider, STORE_PREFIX);

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final StateStore wrapped = ((WrappedStateStore) store).wrapped();
        assertThat(wrapped, instanceOf(CachingKeyValueStore.class));
    }

    @Test
    public void shouldCreateHeadersAwareStoreWithProvidedSupplierAndLoggingEnabled() {
        mockKeyValueStoreSupplier();
        doReturn("headers")
                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);

        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized =
            new MaterializedInternal<>(Materialized.<String, String>as(keyValueStoreSupplier).withCachingDisabled(), nameProvider, STORE_PREFIX);

        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);

        final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped();
        assertThat(innerKeyValueStore.name(), equalTo(store.name()));
        assertThat(store, instanceOf(MeteredTimestampedKeyValueStoreWithHeaders.class));
        assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.class));
    }

    @SuppressWarnings("unchecked")
    private TimestampedKeyValueStore<String, String> getTimestampedStore(
        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized) {
        final KeyValueStoreMaterializer<String, String> materializer = new KeyValueStoreMaterializer<>(materialized);
        materializer.configure(streamsConfig);
        return (TimestampedKeyValueStore<String, String>) materializer.builder().build();
    }