Use a kafka-safe table in FlinkMainIT

Also added the custom classes to the uber JAR for proper classloading

Author: ferenc-csaky

connectors/kafka-safe/pom.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright © 2024 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.datasqrl.flinkrunner</groupId>
+    <artifactId>connectors</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>kafka-safe</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-core</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-streaming-java</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-api-java</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-api-java-bridge</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-kafka</artifactId>
+      <version>${kafka.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.auto.service</groupId>
+      <artifactId>auto-service</artifactId>
+    </dependency>
+  </dependencies>
+</project>

connectors/kafka/src/main/java/com/datasqrl/DeserFailureHandler.java renamed to connectors/kafka-safe/src/main/java/com/datasqrl/DeserFailureHandler.java

@@ -27,9 +27,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class DeserFailureHandler {
+public class DeserFailureHandler implements Serializable {
 
   private static final Logger LOG = LoggerFactory.getLogger(DeserFailureHandler.class);
+  private static final long serialVersionUID = 1L;
 
   private final DeserFailureHandlerType handlerType;
   private final @Nullable DeserFailureProducer producer;

connectors/kafka/src/main/java/com/datasqrl/DeserFailureHandlerOptions.java renamed to connectors/kafka-safe/src/main/java/com/datasqrl/DeserFailureHandlerOptions.java

@@ -30,6 +30,7 @@
 class DeserFailureProducer implements Serializable {
 
   private static final Logger LOG = LoggerFactory.getLogger(DeserFailureProducer.class);
+  private static final long serialVersionUID = 1L;
 
   private final String topic;
   private final Properties producerProps;

connectors/kafka/src/main/java/com/datasqrl/DeserFailureHandlerType.java renamed to connectors/kafka-safe/src/main/java/com/datasqrl/DeserFailureHandlerType.java

@@ -16,31 +16,29 @@
     limitations under the License.
 
 -->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-
-    <modelVersion>4.0.0</modelVersion>
-
-    <parent>
-        <groupId>com.datasqrl.flinkrunner</groupId>
-        <artifactId>flink-sql-runner-parent</artifactId>
-        <version>1.0.0-SNAPSHOT</version>
-    </parent>
-
-    <artifactId>connectors</artifactId>
-    <packaging>pom</packaging>
-
-    <modules>
-        <module>kafka</module>
-    </modules>
-
-    <dependencies>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-base</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
-    </dependencies>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.datasqrl.flinkrunner</groupId>
+    <artifactId>flink-sql-runner-parent</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>connectors</artifactId>
+  <packaging>pom</packaging>
+
+  <modules>
+    <module>kafka-safe</module>
+  </modules>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-base</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+  </dependencies>
 </project>

connectors/kafka/src/main/java/com/datasqrl/DeserFailureProducer.java renamed to connectors/kafka-safe/src/main/java/com/datasqrl/DeserFailureProducer.java

@@ -172,13 +172,12 @@
 
     <dependency>
       <groupId>com.datasqrl.flinkrunner</groupId>
-      <artifactId>system-functions-discovery</artifactId>
+      <artifactId>kafka-safe</artifactId>
       <version>1.0.0-SNAPSHOT</version>
-      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>com.datasqrl.flinkrunner</groupId>
-      <artifactId>kafka</artifactId>
+      <artifactId>system-functions-discovery</artifactId>
       <version>1.0.0-SNAPSHOT</version>
       <scope>provided</scope>
     </dependency>
@@ -380,18 +379,6 @@
               <outputDirectory>${project.build.directory}</outputDirectory>
             </configuration>
           </execution>
-          <execution>
-            <id>copy-kafka-connector</id>
-            <goals>
-              <goal>copy-dependencies</goal>
-            </goals>
-            <phase>process-resources</phase>
-            <configuration>
-              <includeGroupIds>com.datasqrl.flinkrunner</includeGroupIds>
-              <includeArtifactIds>kafka</includeArtifactIds>
-              <outputDirectory>${project.build.directory}</outputDirectory>
-            </configuration>
-          </execution>
         </executions>
       </plugin>
     </plugins>

connectors/kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SafeDynamicKafkaDeserializationSchema.java renamed to connectors/kafka-safe/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SafeDynamicKafkaDeserializationSchema.java

@@ -45,6 +45,8 @@ services:
       - JDBC_URL=jdbc:postgresql://postgres:5432/datasqrl
       - JDBC_USERNAME=postgres
       - JDBC_PASSWORD=postgres
+      - PROPERTIES_BOOTSTRAP_SERVERS=kafka:9092
+      - PROPERTIES_GROUP_ID=mygroupid
       - |
         FLINK_PROPERTIES=
         taskmanager.slot.timeout: 30000ms

connectors/kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SafeKafkaDynamicSource.java renamed to connectors/kafka-safe/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SafeKafkaDynamicSource.java

@@ -296,4 +296,10 @@ void givenUdfPlansScript_whenExecuting_thenSuccess() {
     String planFile = "/opt/flink/usrlib/plans/compiled-plan-udf.json";
     execute("--planfile", planFile, "--udfpath", "/opt/flink/usrlib/udfs/");
   }
+
+  @Test
+  void givenKafkaPlanScript_whenExecuting_thenSuccess() {
+    String planFile = "/opt/flink/usrlib/plans/kafka_plan.json";
+    execute("--planfile", planFile);
+  }
 }