wip

Author: velo

pom.xml

@@ -202,7 +202,14 @@
       <artifactId>flink-s3-fs-hadoop</artifactId>
       <version>${flink.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-exec</artifactId>
+      <version>1.4.0</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
+
   <repositories>
     <repository>
       <id>central</id>
@@ -226,11 +233,6 @@
 
   <build>
     <resources>
-      <resource>
-        <targetPath>${project.basedir}/target</targetPath>
-        <filtering>true</filtering>
-        <directory>src/test/docker</directory>
-      </resource>
       <resource>
         <directory>src/main/resources</directory>
       </resource>

src/test/docker/docker-compose.yml

@@ -38,18 +38,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 +61,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 +95,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,71 @@
+/*
+ * 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(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);
+    }
+  }
+}

src/test/java/com/datasqrl/FlinkMainIT.java

@@ -25,21 +25,36 @@
 import com.nextbreakpoint.flink.client.model.TerminationMode;
 import com.nextbreakpoint.flink.client.model.UploadStatus;
 import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.SneakyThrows;
 import org.apache.flink.shaded.curator5.com.google.common.base.Objects;
 import org.apache.flink.shaded.curator5.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.MethodSource;
 
 class FlinkMainIT extends AbstractITSupport {
 
+  @BeforeEach
+  @AfterEach
+  void terminateJobs() throws Exception {
+    var jobs = client.getJobsOverview();
+    jobs.getJobs().stream()
+        .filter(job -> Objects.equal(job.getState(), JobStatus.RUNNING))
+        .forEach(job -> stopJobs(job.getJid()));
+  }
+
   static Stream<Arguments> sqlScripts() {
     var scripts = List.of("flink.sql", "test_sql.sql");
     var config = List.of(true, false);
@@ -82,6 +97,34 @@ void givenPlansScript_whenExecuting_thenSuccess(String filename, boolean config)
     execute(args.toArray(String[]::new));
   }
 
+  @Test
+  void givenPlanScript_whenCheckPoint_thenResumeSuccessfulySuccess() throws Exception {
+    String planFile = "/opt/flink/usrlib/sqrl/compiled-plan.json";
+    var args = new ArrayList<String>();
+    args.add("--planfile");
+    args.add(planFile);
+    args.add("--config-dir");
+    args.add("/opt/flink/usrlib/config/");
+    var jobResponse = execute(args.toArray(String[]::new));
+
+    TimeUnit.SECONDS.sleep(1);
+
+    Path savePoint = Path.of("/tmp/flink-jar-runner/" + System.currentTimeMillis());
+
+    CommandLineUtil.execute(
+        Path.of("."),
+        "docker exec -t flink-jar-runner-jobmanager-1 bin/flink stop "
+            + jobResponse.getJobid()
+            + " --savepointPath "
+            + savePoint);
+
+    assertThat(savePoint).isNotEmptyDirectory();
+
+    try (Stream<Path> filePathStream = Files.walk(savePoint)) {
+      filePathStream.filter(Files::isRegularFile).forEach(System.out::println);
+    }
+  }
+
   @SneakyThrows
   JarRunResponseBody execute(String... arguments) {
     var jarFile = new File("target/flink-jar-runner.uber.jar");
@@ -91,7 +134,7 @@ JarRunResponseBody execute(String... arguments) {
     assertThat(uploadResponse.getStatus()).isEqualTo(UploadStatus.SUCCESS);
 
     // Step 2: Extract jarId from the response
-    String jarId =
+    var jarId =
         uploadResponse.getFilename().substring(uploadResponse.getFilename().lastIndexOf("/") + 1);
 
     // Step 3: Submit the job
@@ -108,17 +151,20 @@ JarRunResponseBody execute(String... arguments) {
     String jobId = jobResponse.getJobid();
     assertThat(jobId).isNotNull();
 
-    SECONDS.sleep(10);
+    SECONDS.sleep(2);
 
+    return jobResponse;
+  }
+
+  @SneakyThrows
+  private void stopJobs(String jobId) {
     var status = client.getJobStatusInfo(jobId);
     if (Objects.equal(status.getStatus(), JobStatus.RUNNING)) {
       client.cancelJob(jobId, TerminationMode.CANCEL);
     } else {
       JobExceptionsInfoWithHistory exceptions = client.getJobExceptions(jobId, 5, null);
       fail(exceptions.toString());
     }
-
-    return jobResponse;
   }
 
   @ParameterizedTest(name = "{0}")