@Override
    public Processor<K, Change<V>, K, Change<VOut>> get() {
        return new KTableTransformValuesProcessor(transformerSupplier.get());
    }
@Test
    public void shouldTransformValuesWithKeyAndMaterialize() {
        builder
            .addStateStore(storeBuilder(STORE_NAME))
            .table(INPUT_TOPIC, CONSUMED)
            .transformValues(
                new ExclamationValueTransformerSupplier(STORE_NAME, QUERYABLE_NAME),
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as(QUERYABLE_NAME)
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.String()),
                STORE_NAME)
            .toStream()
            .process(capture);

        driver = new TopologyTestDriver(builder.build(), props());
        final TestInputTopic<String, String> inputTopic =
                driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
        inputTopic.pipeInput("A", "a", 5L);
        inputTopic.pipeInput("B", "b", 10L);
        inputTopic.pipeInput("C", null, 15L);

        assertThat(output(), hasItems(new KeyValueTimestamp<>("A", "A->a!", 5),
                new KeyValueTimestamp<>("B", "B->b!", 10),
                new KeyValueTimestamp<>("C", "C->null!", 15)));

        {
            final KeyValueStore<String, String> keyValueStore = driver.getKeyValueStore(QUERYABLE_NAME);
            assertThat(keyValueStore.get("A"), is("A->a!"));
            assertThat(keyValueStore.get("B"), is("B->b!"));
            assertThat(keyValueStore.get("C"), is("C->null!"));
        }
        {
            final KeyValueStore<String, ValueAndTimestamp<String>> keyValueStore = driver.getTimestampedKeyValueStore(QUERYABLE_NAME);
            assertThat(keyValueStore.get("A"), is(ValueAndTimestamp.make("A->a!", 5L)));
            assertThat(keyValueStore.get("B"), is(ValueAndTimestamp.make("B->b!", 10L)));
            assertThat(keyValueStore.get("C"), is(ValueAndTimestamp.make("C->null!", 15L)));
        }
    }