Merge pull request #18 from DataSQRL/upgrade_client_api

Bump flink client library version

Author: velo

pom.xml

@@ -135,8 +135,8 @@
 
     <dependency>
       <groupId>com.nextbreakpoint</groupId>
-      <artifactId>com.nextbreakpoint.flinkclient</artifactId>
-      <version>1.0.4</version>
+      <artifactId>com.nextbreakpoint.flink.client</artifactId>
+      <version>1.1.4</version>
       <scope>test</scope>
     </dependency>
 

src/test/docker/docker-compose.yml

@@ -38,6 +38,7 @@ services:
       - datasqrl_network
 
   jobmanager:
+    user: 1000:1000
     image: flink:${flink-base-image}
     environment:
       - JDBC_URL=jdbc:postgresql://postgres:5432/datasqrl
@@ -70,6 +71,7 @@ services:
       - datasqrl_network
 
   taskmanager:
+    user: 1000:1000
     image: flink:${flink-base-image}
     environment:
       - JDBC_URL=jdbc:postgresql://postgres:5432/datasqrl

src/test/java/com/datasqrl/AbstractITSupport.java

@@ -19,9 +19,11 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.awaitility.Awaitility.await;
 
-import com.nextbreakpoint.flinkclient.api.ApiException;
-import com.nextbreakpoint.flinkclient.api.FlinkApi;
-import com.nextbreakpoint.flinkclient.model.JobIdsWithStatusOverview;
+import com.nextbreakpoint.flink.client.api.ApiException;
+import com.nextbreakpoint.flink.client.api.FlinkApi;
+import com.nextbreakpoint.flink.client.model.JobIdsWithStatusOverview;
+import com.nextbreakpoint.flink.client.model.TerminationMode;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
@@ -38,9 +40,17 @@ static void waitServiceStart() throws ApiException {
     client.getApiClient().setBasePath(serverUrl());
 
     int timeout = (int) TimeUnit.MINUTES.toMillis(2);
-    client.getApiClient().setConnectTimeout(timeout);
-    client.getApiClient().setReadTimeout(timeout);
-    client.getApiClient().setWriteTimeout(timeout);
+    client
+        .getApiClient()
+        .setHttpClient(
+            client
+                .getApiClient()
+                .getHttpClient()
+                .newBuilder()
+                .connectTimeout(Duration.ofMinutes(2))
+                .writeTimeout(Duration.ofMinutes(2))
+                .readTimeout(Duration.ofMinutes(2))
+                .build());
 
     await()
         .atMost(100, SECONDS)
@@ -49,16 +59,16 @@ static void waitServiceStart() throws ApiException {
         .until(
             () -> {
               log.info("Awaiting for custody-api");
-              return client.getJobs() != null;
+              return client.getJobsOverview() != null;
             });
 
-    final JobIdsWithStatusOverview statusOverview = client.getJobs();
+    final JobIdsWithStatusOverview statusOverview = client.getJobIdsWithStatusesOverview();
     statusOverview
         .getJobs()
         .forEach(
             jobIdWithStatus -> {
               try {
-                client.terminateJob(jobIdWithStatus.getId(), "cancel");
+                client.cancelJob(jobIdWithStatus.getId(), TerminationMode.CANCEL);
               } catch (ApiException ignored) {
               }
             });

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

@@ -15,19 +15,23 @@
  */
 package com.datasqrl;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.junit.jupiter.api.Assertions.fail;
 
-import com.nextbreakpoint.flinkclient.model.JarRunResponseBody;
-import com.nextbreakpoint.flinkclient.model.JarUploadResponseBody;
-import com.nextbreakpoint.flinkclient.model.JarUploadResponseBody.StatusEnum;
+import com.nextbreakpoint.flink.client.model.JarRunResponseBody;
+import com.nextbreakpoint.flink.client.model.JobExceptionsInfoWithHistory;
+import com.nextbreakpoint.flink.client.model.JobStatus;
+import com.nextbreakpoint.flink.client.model.TerminationMode;
+import com.nextbreakpoint.flink.client.model.UploadStatus;
 import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 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.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -79,34 +83,42 @@ void givenPlansScript_whenExecuting_thenSuccess(String filename, boolean config)
   }
 
   @SneakyThrows
-  void execute(String... arguments) {
-    File jarFile = new File("target/flink-jar-runner.uber.jar");
+  JarRunResponseBody execute(String... arguments) {
+    var jarFile = new File("target/flink-jar-runner.uber.jar");
 
-    JarUploadResponseBody uploadResponse = client.uploadJar(jarFile);
+    var uploadResponse = client.uploadJar(jarFile);
 
-    assertThat(uploadResponse.getStatus()).isEqualTo(StatusEnum.SUCCESS);
+    assertThat(uploadResponse.getStatus()).isEqualTo(UploadStatus.SUCCESS);
 
     // Step 2: Extract jarId from the response
     String jarId =
         uploadResponse.getFilename().substring(uploadResponse.getFilename().lastIndexOf("/") + 1);
 
     // Step 3: Submit the job
-    assertThatNoException()
-        .as("Running script %s", Arrays.toString(arguments))
-        .isThrownBy(
-            () -> {
-              JarRunResponseBody jobResponse =
-                  client.runJar(
-                      jarId,
-                      null,
-                      null,
-                      null,
-                      Arrays.stream(arguments).collect(Collectors.joining(",")),
-                      null,
-                      1);
-              String jobId = jobResponse.getJobid();
-              assertThat(jobId).isNotNull();
-            });
+    var jobResponse =
+        client.submitJobFromJar(
+            jarId,
+            null,
+            null,
+            null,
+            null,
+            Arrays.stream(arguments).collect(Collectors.joining(",")),
+            null,
+            1);
+    String jobId = jobResponse.getJobid();
+    assertThat(jobId).isNotNull();
+
+    SECONDS.sleep(10);
+
+    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}")