KAFKA-7952: use in memory stores for KTable test (#19218)

Switch to in-memory story for KTable-KTable left-join test.

Reviewers: Matthias J. Sax <[email protected]>

Author: lorcanj

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java

@@ -21,9 +21,11 @@
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.KeyValueTimestamp;
 import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.TestInputTopic;
 import org.apache.kafka.streams.TestOutputTopic;
 import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyConfig;
 import org.apache.kafka.streams.TopologyTestDriver;
 import org.apache.kafka.streams.TopologyTestDriverWrapper;
 import org.apache.kafka.streams.TopologyWrapper;
@@ -35,6 +37,7 @@
 import org.apache.kafka.streams.processor.api.MockProcessorContext;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.test.TestRecord;
 import org.apache.kafka.test.MockApiProcessor;
@@ -70,9 +73,14 @@ public class KTableKTableLeftJoinTest {
     private final Consumed<Integer, String> consumed = Consumed.with(Serdes.Integer(), Serdes.String());
     private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
 
+    private StreamsBuilder createStreamBuilderInMemory() {
+        props.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class.getName());
+        return new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
+    }
+
     @Test
     public void testJoin() {
-        final StreamsBuilder builder = new StreamsBuilder();
+        final StreamsBuilder builder = createStreamBuilderInMemory();
 
         final int[] expectedKeys = new int[] {0, 1, 2, 3};
 
@@ -193,7 +201,7 @@ public void testJoin() {
 
     @Test
     public void testNotSendingOldValue() {
-        final StreamsBuilder builder = new StreamsBuilder();
+        final StreamsBuilder builder = createStreamBuilderInMemory();
 
         final int[] expectedKeys = new int[] {0, 1, 2, 3};
 
@@ -309,7 +317,7 @@ public void testNotSendingOldValue() {
 
     @Test
     public void testSendingOldValue() {
-        final StreamsBuilder builder = new StreamsBuilder();
+        final StreamsBuilder builder = createStreamBuilderInMemory();
 
         final int[] expectedKeys = new int[] {0, 1, 2, 3};
 
@@ -443,7 +451,7 @@ public void shouldNotThrowIllegalStateExceptionWhenMultiCacheEvictions() {
         final String tableSix = "tableSix";
         final String[] inputs = {agg, tableOne, tableTwo, tableThree, tableFour, tableFive, tableSix};
 
-        final StreamsBuilder builder = new StreamsBuilder();
+        final StreamsBuilder builder = createStreamBuilderInMemory();
         final Consumed<Long, String> consumed = Consumed.with(Serdes.Long(), Serdes.String());
         final KTable<Long, String> aggTable = builder
             .table(agg, consumed, Materialized.as(Stores.inMemoryKeyValueStore("agg-base-store")))
@@ -453,30 +461,12 @@ public void shouldNotThrowIllegalStateExceptionWhenMultiCacheEvictions() {
                 MockReducer.STRING_ADDER,
                 Materialized.as(Stores.inMemoryKeyValueStore("agg-store")));
 
-        final KTable<Long, String> one = builder.table(
-            tableOne,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableOne-base-store")));
-        final KTable<Long, String> two = builder.table(
-            tableTwo,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableTwo-base-store")));
-        final KTable<Long, String> three = builder.table(
-            tableThree,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableThree-base-store")));
-        final KTable<Long, String> four = builder.table(
-            tableFour,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableFour-base-store")));
-        final KTable<Long, String> five = builder.table(
-            tableFive,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableFive-base-store")));
-        final KTable<Long, String> six = builder.table(
-            tableSix,
-            consumed,
-            Materialized.as(Stores.inMemoryKeyValueStore("tableSix-base-store")));
+        final KTable<Long, String> one = builder.table(tableOne, consumed);
+        final KTable<Long, String> two = builder.table(tableTwo, consumed);
+        final KTable<Long, String> three = builder.table(tableThree, consumed);
+        final KTable<Long, String> four = builder.table(tableFour, consumed);
+        final KTable<Long, String> five = builder.table(tableFive, consumed);
+        final KTable<Long, String> six = builder.table(tableSix, consumed);
 
         final ValueMapper<String, String> mapper = value -> value.toUpperCase(Locale.ROOT);
 
@@ -515,7 +505,7 @@ public void shouldNotThrowIllegalStateExceptionWhenMultiCacheEvictions() {
 
     @Test
     public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
-        final StreamsBuilder builder = new StreamsBuilder();
+        final StreamsBuilder builder = createStreamBuilderInMemory();
 
         @SuppressWarnings("unchecked")
         final Processor<String, Change<String>, String, Change<Object>> join = new KTableKTableLeftJoin<>(