Test savepoints on compiled plans

Create new integration test that stops job with savepoint, changes
compiled plan, restore jobs and check if changes took effect

Signed-off-by: Marvin Froeder <marvin@datasqrl.com>

Author: velo

pom.xml

@@ -147,6 +147,27 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-exec</artifactId>
+      <version>1.4.0</version>
+      <scope>test</scope>
+    </dependency>
+
+    <!-- JDBI -->
+    <dependency>
+      <groupId>org.jdbi</groupId>
+      <artifactId>jdbi3-core</artifactId>
+      <version>3.45.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.jdbi</groupId>
+      <artifactId>jdbi3-sqlobject</artifactId>
+      <version>3.45.1</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-kafka</artifactId>
@@ -203,6 +224,7 @@
       <version>${flink.version}</version>
     </dependency>
   </dependencies>
+
   <repositories>
     <repository>
       <id>central</id>

src/test/docker/docker-compose.yml

@@ -22,13 +22,15 @@ networks:
 
 services:
   postgres:
-    image: postgres:16
+    image: postgres:17
     environment:
       POSTGRES_USER: postgres
       POSTGRES_PASSWORD: postgres
       POSTGRES_DB: datasqrl
     ports:
       - "5432:5432"
+    volumes:
+      - ./sqrl/postgres-schema.sql:/docker-entrypoint-initdb.d/database-schema.sql:ro
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]
       interval: 1s
@@ -38,18 +40,17 @@ services:
       - datasqrl_network
 
   jobmanager:
-    user: 1000:1000
     image: flink:${flink-base-image}
     environment:
       - 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
         jobmanager.rpc.address: jobmanager
+        state.savepoints-storage: filesystem
+        state.savepoints.dir: file:///tmp/flink-jar-runner
     command: jobmanager
     ports:
       - "8081:8081"  # Flink JobManager REST port
@@ -62,27 +63,32 @@ services:
       - flink_conf:/opt/flink/conf
       - ./:/opt/flink/usrlib/
       - ./datasources/:/datasources/
+      - /tmp:/tmp
     healthcheck:
       test: ["CMD-SHELL", "curl -f http://localhost:8081/ || exit 1"]
       interval: 1s
       timeout: 1s
       retries: 50
+    restart: always
+    deploy:
+      resources:
+        limits:
+          memory: 2g
     networks:
       - datasqrl_network
 
   taskmanager:
-    user: 1000:1000
     image: flink:${flink-base-image}
     environment:
       - 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
         jobmanager.rpc.address: jobmanager
+        state.savepoints-storage: filesystem
+        state.savepoints.dir: file:///tmp/flink-jar-runner
     command: taskmanager
     depends_on:
       jobmanager:
@@ -91,6 +97,12 @@ services:
       - flink_conf:/opt/flink/conf
       - ./:/opt/flink/usrlib/
       - ./datasources/:/datasources/
+      - /tmp:/tmp
+    restart: always
+    deploy:
+      resources:
+        limits:
+          memory: 2g
     networks:
       - datasqrl_network
 

src/test/java/com/datasqrl/CommandLineUtil.java

@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+package com.datasqrl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.ExecuteException;
+import org.apache.commons.exec.ExecuteWatchdog;
+import org.apache.commons.exec.PumpStreamHandler;
+
+@Slf4j
+@UtilityClass
+public class CommandLineUtil {
+
+  public static String execute(String command) throws ExecuteException {
+    return execute(Path.of("."), command);
+  }
+
+  public static String execute(Path workDir, String command) throws ExecuteException {
+    command = command.replaceAll("\\s+", " ").trim();
+
+    log.info("Executing command: {}", command);
+
+    var cmdLine = CommandLine.parse(command);
+
+    var output = new ByteArrayOutputStream();
+    var streamHandler = new PumpStreamHandler(output);
+
+    var executor =
+        DefaultExecutor.builder()
+            .setWorkingDirectory(workDir.toFile())
+            .setExecuteStreamHandler(streamHandler)
+            .get();
+    executor.setExitValue(0);
+
+    var watchdog = ExecuteWatchdog.builder().setTimeout(Duration.ofMinutes(5)).get();
+    executor.setWatchdog(watchdog);
+
+    try {
+      var exitValue =
+          // This call is synchronous and will block until the command completes
+          executor.execute(cmdLine);
+      log.info("Installation completed successfully with exit code: {}", exitValue);
+
+      return new String(output.toByteArray());
+    } catch (IOException e) {
+      var result = new String(output.toByteArray());
+      log.error("Error while executing command:\n{}\noutput:\n{}", command, result);
+      if (e instanceof ExecuteException) {
+        ExecuteException ee = (ExecuteException) e;
+        throw new ExecuteException(result, ee.getExitValue(), ee);
+      }
+      throw new RuntimeException(e);
+    }
+  }
+}