test: Validate proper dynamic table factory wiring (#334)

Author: ferenc-csaky

flink-sql-runner/pom.xml

@@ -346,6 +346,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.github.classgraph</groupId>
+      <artifactId>classgraph</artifactId>
+      <version>${classgraph.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers-junit-jupiter</artifactId>

flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/FactoryAutoServiceTest.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.auto.service.AutoService;
+import io.github.classgraph.AnnotationClassRef;
+import io.github.classgraph.ClassGraph;
+import io.github.classgraph.ClassInfo;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.Factory;
+import org.junit.jupiter.api.Test;
+
+class FactoryAutoServiceTest {
+
+  private static final Set<String> FIRST_PARTY_ARTIFACTS =
+      Set.of(
+          "datagen-connectors",
+          "flexible-csv-format",
+          "flexible-json-format",
+          "flink-sql-runner",
+          "kafka-safe-connector",
+          "postgresql-connector",
+          "stdlib-iceberg");
+
+  @Test
+  void dynamicTableFactoriesAreRegisteredWithAutoService() {
+    try (var scanResult =
+        new ClassGraph()
+            .enableAnnotationInfo()
+            .enableClassInfo()
+            .acceptPackages("com.datasqrl", "org.apache.flink.streaming.connectors.kafka")
+            .scan()) {
+
+      var factories =
+          Stream.concat(
+                  scanResult.getClassesImplementing(DynamicTableSourceFactory.class).stream(),
+                  scanResult.getClassesImplementing(DynamicTableSinkFactory.class).stream())
+              .filter(FactoryAutoServiceTest::isFirstPartyClass)
+              .distinct()
+              .toList();
+
+      assertThat(factories).isNotEmpty();
+
+      var factoriesMissingAutoService =
+          factories.stream()
+              .filter(factory -> !hasAutoServiceFactoryAnnotation(factory))
+              .map(ClassInfo::getName)
+              .toList();
+
+      assertThat(factoriesMissingAutoService)
+          .as("Dynamic table factories missing @AutoService(Factory.class)")
+          .isEmpty();
+    }
+  }
+
+  private static boolean isFirstPartyClass(ClassInfo classInfo) {
+    var classpathElement = classInfo.getClasspathElementURI().toString();
+    return FIRST_PARTY_ARTIFACTS.stream().anyMatch(classpathElement::contains);
+  }
+
+  private static boolean hasAutoServiceFactoryAnnotation(ClassInfo classInfo) {
+    var annotationInfo = classInfo.getAnnotationInfo(AutoService.class.getName());
+    if (annotationInfo == null) {
+      return false;
+    }
+
+    var serviceTypes = annotationInfo.getParameterValues().getValue("value");
+    if (serviceTypes instanceof Object[] array) {
+      return Stream.of(array).anyMatch(FactoryAutoServiceTest::isFactoryClassRef);
+    }
+
+    return isFactoryClassRef(serviceTypes);
+  }
+
+  private static boolean isFactoryClassRef(Object serviceType) {
+    return serviceType instanceof AnnotationClassRef classRef
+        && Factory.class.getName().equals(classRef.getName());
+  }
+}

pom.xml

@@ -100,6 +100,7 @@
     <auto.service.version>1.1.1</auto.service.version>
     <awaitility.version>4.3.0</awaitility.version>
     <aws-msk-iam-auth.version>2.3.6</aws-msk-iam-auth.version>
+    <classgraph.version>4.8.184</classgraph.version>
     <commons-exec.version>1.6.0</commons-exec.version>
     <commons-math3.version>3.6.1</commons-math3.version>
     <feign.version>13.5</feign.version>