Fix for TTD when loading properties from file (#439)

Author: ableegoldman

responsive-test-utils/src/main/java/dev/responsive/kafka/api/ResponsiveTopologyTestDriver.java

@@ -14,6 +14,7 @@
 
 import static dev.responsive.kafka.api.async.internals.AsyncUtils.getAsyncThreadPool;
 import static dev.responsive.kafka.api.config.ResponsiveConfig.ASYNC_THREAD_POOL_SIZE_CONFIG;
+import static dev.responsive.kafka.api.config.ResponsiveConfig.responsiveConfig;
 import static dev.responsive.kafka.internal.stores.TTDRestoreListener.mockRestoreListener;
 
 import dev.responsive.kafka.api.async.internals.AsyncThreadPoolRegistration;
@@ -191,7 +192,7 @@ private static Properties testDriverProps(
         .withMetrics(metrics)
         .withTopologyDescription(topologyDescription);
 
-    AsyncUtils.configuredAsyncThreadPool(ResponsiveConfig.responsiveConfig(baseProps), 1, metrics)
+    AsyncUtils.configuredAsyncThreadPool(responsiveConfig(baseProps), 1, metrics)
         .ifPresent(threadPoolRegistry -> {
           threadPoolRegistry.startNewAsyncThreadPool(Thread.currentThread().getName());
           sessionConfig.withAsyncThreadPoolRegistry(threadPoolRegistry);
@@ -231,7 +232,7 @@ private static MockTime mockTime(final Instant initialWallClockTime) {
   private static Optional<AsyncThreadPoolRegistration> getAsyncThreadPoolRegistration(
       final Properties props
   ) {
-    final int asyncThreadPoolSize = (int) props.getOrDefault(ASYNC_THREAD_POOL_SIZE_CONFIG, 0);
+    final int asyncThreadPoolSize = responsiveConfig(props).getInt(ASYNC_THREAD_POOL_SIZE_CONFIG);
 
     if (asyncThreadPoolSize > 0) {
       final Map<String, Object> configMap = new HashMap<>();

responsive-test-utils/src/test/java/dev/responsive/kafka/api/ResponsiveTopologyTestDriverKeyValueStoreTest.java

@@ -14,6 +14,8 @@
 
 import static dev.responsive.kafka.api.async.AsyncFixedKeyProcessorSupplier.createAsyncProcessorSupplier;
 import static dev.responsive.kafka.testutils.processors.Deduplicator.deduplicatorApp;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
@@ -24,6 +26,7 @@
 import dev.responsive.kafka.api.stores.TtlProvider.TtlDuration;
 import dev.responsive.kafka.internal.stores.SchemaTypes;
 import dev.responsive.kafka.internal.stores.SchemaTypes.KVSchema;
+import java.io.InputStream;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
@@ -45,8 +48,8 @@
 import org.apache.kafka.streams.processor.api.FixedKeyRecord;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.ValueAndTimestamp;
-import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -87,7 +90,7 @@ public void shouldRunWithoutResponsiveConnectionAndNoTtl(final KVSchema type) {
 
       // Then:
       final List<String> outputs = output.readValuesToList();
-      MatcherAssert.assertThat(outputs, Matchers.contains(
+      assertThat(outputs, Matchers.contains(
           "a,100,1,1,alice,CA",
           "c,102,1,1,alice,CA",
           "d,103,3,3,carol,CA"
@@ -145,7 +148,7 @@ public void shouldEnforceKeyBasedTtlByAdvancingStreamTime(final KVSchema type) {
 
       // Then:
       final List<String> outputs = output.readValuesToList();
-      MatcherAssert.assertThat(outputs, Matchers.contains(
+      assertThat(outputs, Matchers.contains(
           "a,100,1,1,alice,CA",
           "d,103,3,3,carol,CA",
           "e,104,1,1,alex,CA",
@@ -157,8 +160,7 @@ public void shouldEnforceKeyBasedTtlByAdvancingStreamTime(final KVSchema type) {
 
   @ParameterizedTest
   @EnumSource(SchemaTypes.KVSchema.class)
-  public void shouldEnforceKeyBasedTtlByAdvancingWallclockTime(final KVSchema type)
-      throws InterruptedException {
+  public void shouldEnforceKeyBasedTtlByAdvancingWallclockTime(final KVSchema type) {
     // Given:
     final Duration defaultTtl = Duration.ofMillis(15);
 
@@ -206,7 +208,7 @@ public void shouldEnforceKeyBasedTtlByAdvancingWallclockTime(final KVSchema type
 
       // Then:
       final List<String> outputs = output.readValuesToList();
-      MatcherAssert.assertThat(outputs, Matchers.containsInAnyOrder(
+      assertThat(outputs, Matchers.containsInAnyOrder(
           "a,100,1,1,alice,CA",
           "d,103,3,3,carol,CA",
           "e,104,1,1,alex,CA",
@@ -263,10 +265,51 @@ public void shouldDeduplicateWithTtlProviderToExpireOldRecords(final KVSchema ty
     }
   }
 
+  @Test
+  public void shouldLoadPropertiesFromFile() throws Exception {
+    final String propsPath = "ttd-app.properties";
+    final InputStream inputStream = getClass().getClassLoader().getResourceAsStream(propsPath);
+
+    if (inputStream == null) {
+      throw new RuntimeException("ttd-app.properties not found under test/resources/");
+    }
+
+    final Properties properties = new Properties();
+    properties.load(inputStream);
+
+    final var params = ResponsiveKeyValueParams.keyValue("store");
+    final Topology topology = deduplicatorApp("input", "output", params);
+    try (final var driver = new ResponsiveTopologyTestDriver(topology, properties, STARTING_TIME)) {
+      final TestInputTopic<String, String> inputTopic = driver.createInputTopic(
+          "input",
+          new StringSerializer(),
+          new StringSerializer(),
+          STARTING_TIME,
+          Duration.ZERO
+      );
+
+      final TestOutputTopic<String, String> output = driver.createOutputTopic(
+          "output", new StringDeserializer(), new StringDeserializer());
+
+      inputTopic.pipeInput("A", "A1");
+      inputTopic.pipeInput("A", "A2");
+      inputTopic.pipeInput("B", "B1");
+      inputTopic.pipeInput("A", "A3");
+
+      final List<String> outputs = output.readValuesToList();
+      assertThat(outputs.size(), equalTo(2));
+      assertThat(outputs, Matchers.containsInAnyOrder(
+          "A1",
+          "B1"
+      ));
+    }
+
+  }
+
   private ResponsiveTopologyTestDriver setupDriver(final Topology topology) {
     final Properties props = new Properties();
-    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
-    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
+    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
+    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
     props.put(ResponsiveConfig.ASYNC_THREAD_POOL_SIZE_CONFIG, 2);
 
     return new ResponsiveTopologyTestDriver(topology, props, STARTING_TIME);

responsive-test-utils/src/test/resources/ttd-app.properties

@@ -0,0 +1,4 @@
+default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+
+responsive.async.thread.pool.size=1