public void enableSendingOldValues() {
        this.sendOldValues = true;
        this.queryableName = storeName;
    }
@Test
    public void testSendingOldValue() {
        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);
        table1.enableSendingOldValues(true);
        assertTrue(table1.sendingOldValueEnabled());

        final MockApiProcessorSupplier<String, Integer, Void, Void> supplier = new MockApiProcessorSupplier<>();
        final Topology topology = builder.build().addProcessor("proc1", supplier, table1.name);

        try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
            final TestInputTopic<String, String> inputTopic1 =
                driver.createInputTopic(
                    topic1,
                    new StringSerializer(),
                    new StringSerializer(),
                    Instant.ofEpochMilli(0L),
                    Duration.ZERO
                );
            final MockApiProcessor<String, Integer, Void, Void> proc1 = supplier.theCapturedProcessor();

            inputTopic1.pipeInput("A", "01", 10L);
            inputTopic1.pipeInput("B", "01", 20L);
            inputTopic1.pipeInput("C", "01", 15L);
            proc1.checkAndClearProcessResult(
                new KeyValueTimestamp<>("A", new Change<>("01", null), 10),
                new KeyValueTimestamp<>("B", new Change<>("01", null), 20),
                new KeyValueTimestamp<>("C", new Change<>("01", null), 15)
            );

            inputTopic1.pipeInput("A", "02", 8L);
            inputTopic1.pipeInput("B", "02", 22L);
            proc1.checkAndClearProcessResult(
                new KeyValueTimestamp<>("A", new Change<>("02", "01"), 8),
                new KeyValueTimestamp<>("B", new Change<>("02", "01"), 22)
            );

            inputTopic1.pipeInput("A", "03", 12L);
            proc1.checkAndClearProcessResult(
                new KeyValueTimestamp<>("A", new Change<>("03", "02"), 12)
            );

            inputTopic1.pipeInput("A", null, 15L);
            inputTopic1.pipeInput("B", null, 20L);
            proc1.checkAndClearProcessResult(
                new KeyValueTimestamp<>("A", new Change<>(null, "03"), 15),
                new KeyValueTimestamp<>("B", new Change<>(null, "02"), 20)
            );
        }
    }