feat: DH-22264, DH-22265: add KafkaTools integration test

This adds KafkaTools integration testing via TestContainers.

There are four different Kafka images we are testing against: `cp-kafka`, `kafka`, `kafka-native`, and `redpanda`.

This includes a test case for DH-22265 / deephaven#7887

Author: devinrsmith

docker/registry/cp-kafka/build.gradle

@@ -0,0 +1,3 @@
+plugins {
+    id 'io.deephaven.project.register'
+}

docker/registry/cp-kafka/gradle.properties

@@ -0,0 +1,6 @@
+io.deephaven.project.ProjectType=DOCKER_REGISTRY
+# Even though the image is expected to be bumped from latest, we should correlate that with the specific version number
+# for easier reference.
+# 8.2.0
+deephaven.registry.imageName=confluentinc/cp-kafka:latest
+deephaven.registry.imageId=confluentinc/cp-kafka@sha256:acbbf674f2ed40e5d0a8ca51beb0f00692c866fc22b5ce06f8cadbdc54cd4436

docker/registry/kafka-native/build.gradle

@@ -0,0 +1,3 @@
+plugins {
+    id 'io.deephaven.project.register'
+}

docker/registry/kafka-native/gradle.properties

@@ -0,0 +1,6 @@
+io.deephaven.project.ProjectType=DOCKER_REGISTRY
+# Even though the image is expected to be bumped from latest, we should correlate that with the specific version number
+# for easier reference.
+# 4.2.0
+deephaven.registry.imageName=apache/kafka-native:latest
+deephaven.registry.imageId=apache/kafka-native@sha256:777f2dddec6970003f1f27922a8c317d87140567b0537e801d35669ad9a81faf

docker/registry/kafka/build.gradle

@@ -0,0 +1,3 @@
+plugins {
+    id 'io.deephaven.project.register'
+}

docker/registry/kafka/gradle.properties

@@ -0,0 +1,6 @@
+io.deephaven.project.ProjectType=DOCKER_REGISTRY
+# Even though the image is expected to be bumped from latest, we should correlate that with the specific version number
+# for easier reference.
+# 4.2.0
+deephaven.registry.imageName=apache/kafka:latest
+deephaven.registry.imageId=apache/kafka@sha256:9516fb7634bad307d17c33b589fde9023003b0cb761374f500002b980a3149b9

docker/registry/redpanda/build.gradle

@@ -0,0 +1,3 @@
+plugins {
+    id 'io.deephaven.project.register'
+}

docker/registry/redpanda/gradle.properties

@@ -0,0 +1,6 @@
+io.deephaven.project.ProjectType=DOCKER_REGISTRY
+# Even though the image is expected to be bumped from latest, we should correlate that with the specific version number
+# for easier reference.
+# v26.1.2
+deephaven.registry.imageName=docker.redpanda.com/redpandadata/redpanda:latest
+deephaven.registry.imageId=docker.redpanda.com/redpandadata/redpanda@sha256:65a7a93449f113a2941e0e8a9cd7eb1ed3ba153547b6b2b47f11796e9f754345

extensions/kafka/build.gradle

@@ -3,6 +3,11 @@ plugins {
     id 'io.deephaven.project.register'
 }
 
+evaluationDependsOn Docker.registryProject('cp-kafka')
+evaluationDependsOn Docker.registryProject('kafka')
+evaluationDependsOn Docker.registryProject('kafka-native')
+evaluationDependsOn Docker.registryProject('redpanda')
+
 description 'Kafka: Integrating Engine tables with Kafka'
 
 dependencies {
@@ -40,8 +45,17 @@ dependencies {
     compileOnly project(':util-immutables')
     annotationProcessor libs.immutables.value
 
+    testImplementation libs.testcontainers
+    testImplementation libs.testcontainers.kafka
+    testImplementation libs.testcontainers.redpanda
+
+    testImplementation project(':engine-test-utils')
+
+    testImplementation project(':extensions-json-jackson')
+
     testImplementation platform(libs.junit.bom)
     testImplementation libs.junit.jupiter
+    testImplementation libs.junit.jupiter.params
     testRuntimeOnly libs.junit.jupiter.engine
     testRuntimeOnly libs.junit.platform.launcher
 
@@ -54,5 +68,28 @@ dependencies {
 }
 
 test {
-    useJUnitPlatform()
+    useJUnitPlatform {
+        excludeTags("testcontainers")
+    }
+}
+
+tasks.register('testOutOfBand', Test) {
+    useJUnitPlatform {
+        includeTags("testcontainers")
+    }
+
+    dependsOn Docker.registryTask(project, 'cp-kafka')
+    systemProperty 'testcontainers.cp-kafka.image', Docker.localImageName('cp-kafka')
+
+    dependsOn Docker.registryTask(project, 'kafka')
+    systemProperty 'testcontainers.kafka.image', Docker.localImageName('kafka')
+
+    dependsOn Docker.registryTask(project, 'kafka-native')
+    systemProperty 'testcontainers.kafka-native.image', Docker.localImageName('kafka-native')
+
+    dependsOn Docker.registryTask(project, 'redpanda')
+    systemProperty 'testcontainers.redpanda.image', Docker.localImageName('redpanda')
+
+    // Kafka client logging is very verbose; tune it down to only log warn or more severe
+    systemProperty 'org.slf4j.simpleLogger.log.org.apache.kafka', 'warn'
 }

extensions/kafka/src/main/java/io/deephaven/kafka/KafkaTools.java

@@ -113,6 +113,7 @@
 import org.immutables.value.Value.Parameter;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -1034,7 +1035,20 @@ public static Table consumeToTable(
             @NotNull final Consume.KeyOrValueSpec keySpec,
             @NotNull final Consume.KeyOrValueSpec valueSpec,
             @NotNull final TableType tableType) {
-        final MutableObject<Table> resultHolder = new MutableObject<>();
+        return consumeToTableAndAdapter(kafkaProperties, topic, partitionFilter, partitionToInitialOffset, keySpec,
+                valueSpec, tableType).table();
+    }
+
+    @VisibleForTesting
+    static TableAndAdapter consumeToTableAndAdapter(
+            @NotNull final Properties kafkaProperties,
+            @NotNull final String topic,
+            @NotNull final IntPredicate partitionFilter,
+            @NotNull final IntToLongFunction partitionToInitialOffset,
+            @NotNull final Consume.KeyOrValueSpec keySpec,
+            @NotNull final Consume.KeyOrValueSpec valueSpec,
+            @NotNull final TableType tableType) {
+        final MutableObject<TableAndAdapter> resultHolder = new MutableObject<>();
         final ExecutionContext enclosingExecutionContext = ExecutionContext.getContext();
         final LivenessManager enclosingLivenessManager = LivenessScopeStack.peek();
 
@@ -1052,14 +1066,35 @@ public static Table consumeToTable(
                         final Table blinkTable = streamToBlinkTableAdapter.table();
                         final Table result = tableType.walk(new BlinkTableOperation(blinkTable));
                         enclosingLivenessManager.manage(result);
-                        resultHolder.setValue(result);
+                        // Note: not adding streamToBlinkTableAdapter to liveness manager; liveness is expected to be
+                        // retained through the result table. This is only relevant for testing.
+                        resultHolder.setValue(new TableAndAdapter(result, streamToBlinkTableAdapter));
                     }
                 };
 
         consume(kafkaProperties, topic, partitionFilter,
                 InitialOffsetLookup.adapt(partitionToInitialOffset), keySpec, valueSpec,
                 StreamConsumerRegistrarProvider.single(registrar), null);
-        return resultHolder.getValue();
+        return resultHolder.get();
+    }
+
+    @VisibleForTesting
+    static class TableAndAdapter {
+        private final Table table;
+        private final StreamToBlinkTableAdapter adapter;
+
+        private TableAndAdapter(Table table, StreamToBlinkTableAdapter adapter) {
+            this.table = Objects.requireNonNull(table);
+            this.adapter = Objects.requireNonNull(adapter);
+        }
+
+        public Table table() {
+            return table;
+        }
+
+        public StreamToBlinkTableAdapter adapter() {
+            return adapter;
+        }
     }
 
     /**