async transformer improvements (#62)

laxmanchekka · web-flow · commit f58941e290e2 · 2023-06-21T14:59:08.000+05:30
diff --git a/kafka-streams-framework/build.gradle.kts b/kafka-streams-framework/build.gradle.kts
@@ -11,6 +11,11 @@ tasks.test {
 }
 
 dependencies {
+  constraints {
+    implementation("org.xerial.snappy:snappy-java:1.1.10.1") {
+      because("[https://nvd.nist.gov/vuln/detail/CVE-2023-34455] in 'org.apache.kafka:kafka-clients:*' > 'org.xerial.snappy:snappy-java:1.1.8.2'")
+    }
+  }
   annotationProcessor("org.projectlombok:lombok:1.18.26")
   compileOnly("org.projectlombok:lombok:1.18.26")
 
diff --git a/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/AsyncTransformer.java b/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/AsyncTransformer.java
@@ -28,7 +28,13 @@ public AsyncTransformer(
     this.executor = executorSupplier.get();
     this.pendingFutures = new ArrayBlockingQueue<>(asyncTransformerConfig.getMaxBatchSize());
     this.rateLimiter =
-        RateLimiter.create(1.0 / asyncTransformerConfig.getCommitIntervalMs().toSeconds());
+        RateLimiter.create(1000.0 / asyncTransformerConfig.getCommitIntervalMs().toMillis());
+    log.info(
+        "async transformer config. maxBatchSize: {}, commit rate: {}",
+        this.pendingFutures.remainingCapacity(),
+        this.rateLimiter.getRate());
+    // warmup to prevent commit on first message
+    rateLimiter.tryAcquire();
   }
 
   @Override
@@ -72,7 +78,14 @@ private void processResults() {
       // makes sure transformation is complete
       future.join();
       // another join is needed to make sure downstream forward is also complete
-      future.thenAccept(result -> result.forEach(kv -> context.forward(kv.key, kv.value))).join();
+      future
+          .thenAccept(
+              result -> {
+                if (result != null) {
+                  result.forEach(kv -> context.forward(kv.key, kv.value));
+                }
+              })
+          .join();
     }
     // commit once per batch
     context.commit();
diff --git a/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/AsyncTransformerConfig.java b/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/AsyncTransformerConfig.java
@@ -35,6 +35,7 @@ public static AsyncTransformerConfig buildWith(Config config, String transformer
               : DEFAULT_ASYNC_TRANSFORMER_COMMIT_INTERVAL;
       return new AsyncTransformerConfig(batchSize, commitInterval);
     }
+
     return new AsyncTransformerConfig(
         DEFAULT_ASYNC_TRANSFORMER_BATCH_SIZE, DEFAULT_ASYNC_TRANSFORMER_COMMIT_INTERVAL);
   }
diff --git a/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/ExecutorFactory.java b/kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/async/ExecutorFactory.java
@@ -6,9 +6,11 @@
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.typesafe.config.Config;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.function.Supplier;
+import org.hypertrace.core.serviceframework.metrics.PlatformMetricsRegistry;
 
 public class ExecutorFactory {
   private static Executor executor;
@@ -20,12 +22,20 @@ public static synchronized Supplier<Executor> getExecutorSupplier(Config config)
               ? config.getInt(ASYNC_EXECUTOR_POOL_SIZE_KEY)
               : DEFAULT_ASYNC_EXECUTOR_POOL_SIZE;
 
-      ThreadFactory threadFactory =
-          new ThreadFactoryBuilder()
-              .setNameFormat("kafka-streams-async-worker-%d")
-              .setDaemon(true)
-              .build();
-      executor = Executors.newFixedThreadPool(poolSize, threadFactory);
+      if (poolSize > 0) {
+        ThreadFactory threadFactory =
+            new ThreadFactoryBuilder()
+                .setNameFormat("kafka-streams-async-worker-%d")
+                .setDaemon(true)
+                .build();
+        ExecutorService executorSvc = Executors.newFixedThreadPool(poolSize, threadFactory);
+        PlatformMetricsRegistry.monitorExecutorService(
+            "kafka-streams-async-pool", executorSvc, null);
+        executor = executorSvc;
+      } else {
+        // direct/sync execution when pool size is explicitly configured to a value <= 0
+        executor = Runnable::run;
+      }
     }
     return () -> executor;
   }
diff --git a/kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/SampleAsyncApp.java b/kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/SampleAsyncApp.java
@@ -67,8 +67,8 @@ public String getServiceName() {
 class SlowTransformer extends AsyncTransformer<String, String, String, String> {
 
   public SlowTransformer(
-      Supplier<Executor> executorSupplier, AsyncTransformerConfig asyncTransformerConfigBuilder) {
-    super(executorSupplier, asyncTransformerConfigBuilder);
+      Supplier<Executor> executorSupplier, AsyncTransformerConfig asyncTransformerConfig) {
+    super(executorSupplier, asyncTransformerConfig);
   }
 
   @Override
@@ -79,6 +79,9 @@ protected void doInit(Map<String, Object> appConfigs) {
   @SneakyThrows
   @Override
   public List<KeyValue<String, String>> asyncTransform(String key, String value) {
+    if (!key.startsWith("key")) {
+      return null;
+    }
     log.info("transforming - key: {}, value: {}", key, value);
     Thread.sleep(25);
     return List.of(KeyValue.pair("out:" + key, "out:" + value));
diff --git a/kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/SampleAsyncAppTest.java b/kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/SampleAsyncAppTest.java
@@ -45,6 +45,13 @@ public void setup() {
     streamsConfig.putAll(sampleApp.streamsConfig);
 
     td = new TopologyTestDriver(sampleApp.topology, streamsConfig);
+
+    inputTopic =
+        td.createInputTopic(
+            SampleApp.INPUT_TOPIC, Serdes.String().serializer(), Serdes.String().serializer());
+    outputTopic =
+        td.createOutputTopic(
+            SampleApp.OUTPUT_TOPIC, Serdes.String().deserializer(), Serdes.String().deserializer());
   }
 
   @AfterEach
@@ -55,22 +62,17 @@ public void tearDown() {
   @SneakyThrows
   @Test
   public void asyncTransformationTest() {
-    inputTopic =
-        td.createInputTopic(
-            SampleApp.INPUT_TOPIC, Serdes.String().serializer(), Serdes.String().serializer());
-    outputTopic =
-        td.createOutputTopic(
-            SampleApp.OUTPUT_TOPIC, Serdes.String().deserializer(), Serdes.String().deserializer());
-
     assertThat(outputTopic.isEmpty(), is(true));
 
     int batchSize = 1000;
     for (int i = 1; i <= batchSize; i++) {
       inputTopic.pipeInput("key-" + i, "value-" + i);
     }
 
+    // sleep time has to be bigger than the commit interval configured for the test
+    // otw, test will be flaky
     Thread.sleep(1000);
-    inputTopic.pipeInput("final", "final");
+    inputTopic.pipeInput("key-final", "value-final");
     // test ordered processing. mandatory requirement.
     for (int i = 1; i <= batchSize; i++) {
       assertThat(outputTopic.readValue(), endsWith("value-" + i));
@@ -82,6 +84,21 @@ public void asyncTransformationTest() {
     assertThat(outputTopic.isEmpty(), is(true));
   }
 
+  @SneakyThrows
+  @Test
+  public void testUnderlyingReturnsNull() {
+    assertThat(outputTopic.isEmpty(), is(true));
+    // null should be handled without erring out.
+    inputTopic.pipeInput("discarded-key", "discarded-value");
+
+    Thread.sleep(1000);
+    inputTopic.pipeInput("key-final", "value-final");
+
+    // read the final record
+    outputTopic.readKeyValue();
+    assertThat(outputTopic.isEmpty(), is(true));
+  }
+
   @Test
   public void baseStreamsConfigTest() {
     Map<String, Object> baseStreamsConfig = sampleApp.getBaseStreamsConfig();
diff --git a/kafka-streams-framework/src/test/resources/configs/sample-kafka-streams-service/application.conf b/kafka-streams-framework/src/test/resources/configs/sample-kafka-streams-service/application.conf
@@ -12,8 +12,8 @@ kafka.streams.config = {
     }
     transformers {
       slow.transformer = {
-        maxBatchSize = 5120
-        commitIntervalMs = 10
+        maxBatchSize = 32
+        commitIntervalMs = 100
       }
     }
   }
diff --git a/kafka-streams-partitioners/avro-partitioners/build.gradle.kts b/kafka-streams-partitioners/avro-partitioners/build.gradle.kts
@@ -11,6 +11,14 @@ tasks.test {
 }
 
 dependencies {
+  constraints {
+    implementation("com.fasterxml.jackson.core:jackson-databind:2.15.2")
+
+    implementation("org.xerial.snappy:snappy-java:1.1.10.1") {
+      because("[https://nvd.nist.gov/vuln/detail/CVE-2023-34455] in 'org.apache.kafka:kafka-clients:*' > 'org.xerial.snappy:snappy-java:1.1.8.2'")
+    }
+  }
+
   annotationProcessor("org.projectlombok:lombok:1.18.24")
   compileOnly("org.projectlombok:lombok:1.18.24")
 
@@ -21,10 +29,6 @@ dependencies {
   implementation("org.apache.kafka:kafka-streams:7.2.1-ccs")
   implementation("org.slf4j:slf4j-api:1.7.36")
 
-  constraints {
-    implementation("com.fasterxml.jackson.core:jackson-databind:2.15.2")
-  }
-
   testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
   testImplementation("org.junit-pioneer:junit-pioneer:1.7.1")
   testImplementation("org.mockito:mockito-core:4.5.1")
diff --git a/kafka-streams-partitioners/weighted-group-partitioner/build.gradle.kts b/kafka-streams-partitioners/weighted-group-partitioner/build.gradle.kts
@@ -10,6 +10,14 @@ tasks.test {
 }
 
 dependencies {
+  constraints {
+    implementation("com.fasterxml.jackson.core:jackson-databind:2.15.2")
+
+    implementation("org.xerial.snappy:snappy-java:1.1.10.1") {
+      because("[https://nvd.nist.gov/vuln/detail/CVE-2023-34455] in 'org.apache.kafka:kafka-clients:*' > 'org.xerial.snappy:snappy-java:1.1.8.2'")
+    }
+  }
+
   annotationProcessor("org.projectlombok:lombok:1.18.24")
   compileOnly("org.projectlombok:lombok:1.18.24")
 
diff --git a/kafka-streams-serdes/build.gradle.kts b/kafka-streams-serdes/build.gradle.kts
@@ -11,14 +11,18 @@ tasks.test {
 }
 
 dependencies {
+  constraints {
+    implementation("com.fasterxml.jackson.core:jackson-databind:2.15.2")
+
+    implementation("org.xerial.snappy:snappy-java:1.1.10.1") {
+      because("[https://nvd.nist.gov/vuln/detail/CVE-2023-34455] in 'org.apache.kafka:kafka-clients:*' > 'org.xerial.snappy:snappy-java:1.1.8.2'")
+    }
+  }
+
   api("org.apache.kafka:kafka-clients:7.2.1-ccs")
   api("org.apache.avro:avro:1.11.1")
 
   testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
-
-  constraints {
-    implementation("com.fasterxml.jackson.core:jackson-databind:2.15.2")
-  }
 }
 
 // Disabling compatibility check for the test avro definitions.
diff --git a/owasp-suppressions.xml b/owasp-suppressions.xml
@@ -2,9 +2,26 @@
 <suppressions xmlns="https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.3.xsd">
     <suppress>
         <notes><![CDATA[
-   Any hypertrace core dep
+   This is a false alert.
+   Hypertrace grpc packages are falsely identified as old versions of open source grpc packages.
    ]]></notes>
-        <packageUrl regex="true">^pkg:maven/org\.hypertrace\.core\..*@.*$</packageUrl>
+        <packageUrl regex="true">^pkg:maven/org\.hypertrace\.core\.grpcutils/grpc-client-utils@.*$</packageUrl>
         <cpe>cpe:/a:grpc:grpc</cpe>
     </suppress>
+    <suppress>
+        <notes><![CDATA[
+   This is a false alert.
+   Hypertrace grpc packages are falsely identified as old versions of open source grpc packages.
+   ]]></notes>
+        <packageUrl regex="true">^pkg:maven/org\.hypertrace\.core\.grpcutils/grpc-context-utils@.*$</packageUrl>
+        <cpe>cpe:/a:grpc:grpc</cpe>
+    </suppress>
+    <suppress>
+        <notes><![CDATA[
+   file name: jackson-databind-2.15.2.jar
+   Excluding the disputed vulnerability (CVE-2023-35116)
+   ]]></notes>
+        <packageUrl regex="true">^pkg:maven/com\.fasterxml\.jackson\.core/jackson\-databind@.*$</packageUrl>
+        <vulnerabilityName>CVE-2023-35116</vulnerabilityName>
+    </suppress>
 </suppressions>

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ public static AsyncTransformerConfig buildWith(Config config, String transformer`
`35`	`35`	`: DEFAULT_ASYNC_TRANSFORMER_COMMIT_INTERVAL;`
`36`	`36`	`return new AsyncTransformerConfig(batchSize, commitInterval);`
`37`	`37`	`}`
	`38`	`+`
`38`	`39`	`return new AsyncTransformerConfig(`
`39`	`40`	`DEFAULT_ASYNC_TRANSFORMER_BATCH_SIZE, DEFAULT_ASYNC_TRANSFORMER_COMMIT_INTERVAL);`
`40`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,8 +12,8 @@ kafka.streams.config = {`
`12`	`12`	`}`
`13`	`13`	`transformers {`
`14`	`14`	`slow.transformer = {`
`15`		`- maxBatchSize = 5120`
`16`		`- commitIntervalMs = 10`
	`15`	`+ maxBatchSize = 32`
	`16`	`+ commitIntervalMs = 100`
`17`	`17`	`}`
`18`	`18`	`}`
`19`	`19`	`}`