@Override
    public Processor<KIn, VIn, KIn, Change<VIn>> get() {
        return new KTableSourceProcessor();
    }
@Test
    public void testValueGetter() {
        final StreamsBuilder builder = new StreamsBuilder();
        final String topic1 = "topic1";

        @SuppressWarnings("unchecked")
        final KTableImpl<String, String, String> table1 =
            (KTableImpl<String, String, String>) builder.table(topic1, stringConsumed, Materialized.as("store"));

        final Topology topology = builder.build();
        final KTableValueGetterSupplier<String, String> getterSupplier1 = table1.valueGetterSupplier();

        final InternalTopologyBuilder topologyBuilder = TopologyWrapper.getInternalTopologyBuilder(topology);
        topologyBuilder.connectProcessorAndStateStores(table1.name, getterSupplier1.storeNames());

        try (final TopologyTestDriverWrapper driver = new TopologyTestDriverWrapper(builder.build(), props)) {
            final TestInputTopic<String, String> inputTopic1 =
                driver.createInputTopic(
                    topic1,
                    new StringSerializer(),
                    new StringSerializer(),
                    Instant.ofEpochMilli(0L),
                    Duration.ZERO
                );
            final KTableValueGetter<String, String> getter1 = getterSupplier1.get();
            getter1.init(driver.setCurrentNodeForProcessorContext(table1.name));

            inputTopic1.pipeInput("A", "01", 10L);
            inputTopic1.pipeInput("B", "01", 20L);
            inputTopic1.pipeInput("C", "01", 15L);

            assertEquals(ValueAndTimestamp.make("01", 10L), getter1.get("A"));
            assertEquals(ValueAndTimestamp.make("01", 20L), getter1.get("B"));
            assertEquals(ValueAndTimestamp.make("01", 15L), getter1.get("C"));

            inputTopic1.pipeInput("A", "02", 30L);
            inputTopic1.pipeInput("B", "02", 5L);

            assertEquals(ValueAndTimestamp.make("02", 30L), getter1.get("A"));
            assertEquals(ValueAndTimestamp.make("02", 5L), getter1.get("B"));
            assertEquals(ValueAndTimestamp.make("01", 15L), getter1.get("C"));

            inputTopic1.pipeInput("A", "03", 29L);

            assertEquals(ValueAndTimestamp.make("03", 29L), getter1.get("A"));
            assertEquals(ValueAndTimestamp.make("02", 5L), getter1.get("B"));
            assertEquals(ValueAndTimestamp.make("01", 15L), getter1.get("C"));

            inputTopic1.pipeInput("A", null, 50L);
            inputTopic1.pipeInput("B", null, 3L);

            assertNull(getter1.get("A"));
            assertNull(getter1.get("B"));
            assertEquals(ValueAndTimestamp.make("01", 15L), getter1.get("C"));
        }
    }