public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
                                                                  final String sourceName,
                                                                  final TimestampExtractor timestampExtractor,
                                                                  final Deserializer<KIn> keyDeserializer,
                                                                  final Deserializer<VIn> valueDeserializer,
                                                                  final String topic,
                                                                  final String processorName,
                                                                  final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
        storeBuilder.withLoggingDisabled();

        internalTopologyBuilder.addSource(AutoOffsetReset.EARLIEST, sourceName, timestampExtractor, keyDeserializer, valueDeserializer, topic);
        internalTopologyBuilder.addProcessor(processorName, stateUpdateSupplier, sourceName);
        internalTopologyBuilder.addStateStore(storeBuilder, processorName);
        internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topic);

        return this;
    }
@Test
    public void readOnlyStateStoresShouldNotLog() {
        final String sourceName = "source";
        final String storeName = "store";
        final String topicName = "topic";
        final String processorName = "processor";

        final KeyValueStoreBuilder<?, ?> storeBuilder = mock(KeyValueStoreBuilder.class);
        when(storeBuilder.name()).thenReturn(storeName);
        topology.addReadOnlyStateStore(
                storeBuilder,
                sourceName,
                null,
                null,
                null,
                topicName,
                processorName,
                new MockProcessorSupplier<>());

        final StoreFactory stateStoreFactory = topology.internalTopologyBuilder.stateStores().get(storeName);
        assertThat(stateStoreFactory.loggingEnabled(), equalTo(false));
    }