@Override
    public KTable<K, V> toTable() {
        return toTable(NamedInternal.empty(), Materialized.with(keySerde, valueSerde));
    }
@Test
    public void shouldMaterializeKTableFromKStream() {
        final Consumed<String, String> consumed = Consumed.with(Serdes.String(), Serdes.String());

        final StreamsBuilder builder = new StreamsBuilder();
        final String storeName = "store";

        final String input = "input";
        builder.stream(input, consumed)
            .toTable(Materialized.as(Stores.inMemoryKeyValueStore(storeName)));

        final Topology topology = builder.build();

        final String topologyDescription = topology.describe().toString();
        assertThat(
            topologyDescription,
            equalTo("Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
                "      --> KSTREAM-TOTABLE-0000000001\n" +
                "    Processor: KSTREAM-TOTABLE-0000000001 (stores: [store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n")
        );

        try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
            final TestInputTopic<String, String> inputTopic =
                driver.createInputTopic(input, new StringSerializer(), new StringSerializer());
            final KeyValueStore<String, String> store = driver.getKeyValueStore(storeName);

            inputTopic.pipeInput("A", "01");
            inputTopic.pipeInput("B", "02");
            inputTopic.pipeInput("A", "03");
            final Map<String, String> expectedStore = mkMap(mkEntry("A", "03"), mkEntry("B", "02"));

            assertThat(asMap(store), is(expectedStore));
        }
    }