feat: Update Flink to 2.3.0 (#364)

Author: ferenc-csaky

flink-sql-runner/src/main/docker/Dockerfile

@@ -14,7 +14,7 @@
 # limitations under the License.
 #
 
-FROM apache/flink:${flink-base-image}
+FROM flink:${flink-base-image}
 
 ARG ASYNC_PROFILER_VERSION=4.4
 ARG TARGETARCH

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

@@ -60,14 +60,14 @@ public class AbstractITSupport {
 
   @Container
   protected static final GenericContainer<?> postgresContainer =
-      new PostgreSQLContainer(DockerImageName.parse("postgres:17"))
+      new PostgreSQLContainer(DockerImageName.parse("postgres:18"))
           .withNetwork(sharedNetwork)
           .withNetworkAliases("postgres")
           .withDatabaseName("datasqrl")
           .withUsername("postgres")
           .withPassword("postgres")
           .withCopyFileToContainer(
-              MountableFile.forClasspathResource("sqrl/postgres-schema.sql"),
+              MountableFile.forClasspathResource("savepoint-test/postgres-schema.sql"),
               "/docker-entrypoint-initdb.d/init.sql");
 
   @Container
@@ -106,13 +106,14 @@ protected void init() throws Exception {
         new GenericContainer<>(DockerImageName.parse("flink-sql-runner"))
             .withNetwork(sharedNetwork)
             .withExposedPorts(FLINK_PORT)
-            .withEnv("JDBC_URL", "jdbc:postgresql://postgres:5432/datasqrl")
-            .withEnv("JDBC_USERNAME", "postgres")
-            .withEnv("JDBC_PASSWORD", "postgres")
+            .withEnv("POSTGRES_AUTHORITY", "postgres:5432/datasqrl")
+            .withEnv("POSTGRES_USERNAME", "postgres")
+            .withEnv("POSTGRES_PASSWORD", "postgres")
             .withEnv("REDPANDA_PORT", String.valueOf(REDPANDA_PORT))
             .withFileSystemBind("target/test-classes/plans", "/it/planfile", BindMode.READ_ONLY)
             .withFileSystemBind("target/test-classes/sql", "/it/sqlfile", BindMode.READ_ONLY)
-            .withFileSystemBind("target/test-classes/sqrl", "/it/sqrl", BindMode.READ_ONLY)
+            .withFileSystemBind(
+                "target/test-classes/savepoint-test", "/it/savepoint-test", BindMode.READ_ONLY)
             .withFileSystemBind("target/test-classes/config", "/it/config", BindMode.READ_ONLY)
             .withFileSystemBind("target/test-classes/udfs", "/it/udfs", BindMode.READ_ONLY)
             .withCommand("bash", "-c", "bin/start-cluster.sh && tail -f /dev/null")

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

@@ -26,7 +26,7 @@
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
-import java.util.ArrayList;
+import java.util.List;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
@@ -56,12 +56,9 @@ void givenPlanScript_whenSavepoint_thenResumeFromSavepointSuccessfully() throws
     // change the compilation plan
     updateCompiledPlan(3);
 
-    var args = new ArrayList<String>();
-    var planFile = "/it/sqrl/compiled-plan.json";
-    args.add("--planfile");
-    args.add(planFile);
-    args.add("--config-dir");
-    args.add("/it/config/");
+    var args =
+        List.of(
+            "--planfile", "/it/savepoint-test/compiled-plan.json", "--config-dir", "/it/config/");
 
     var jobId = flinkRun(args);
     assertJobIsRunning(jobId);
@@ -149,9 +146,11 @@ public void logException(StatementContext ctx, SQLException ex) {
 
   @SneakyThrows
   private void updateCompiledPlan(int newValue) {
-    var contents = Files.readString(Path.of("src/test/resources/sqrl/compiled-plan.json"), UTF_8);
+    var contents =
+        Files.readString(Path.of("src/test/resources/savepoint-test/compiled-plan.json"), UTF_8);
     contents = contents.replace("\"value\" : 2", "\"value\" : " + newValue);
-    Files.writeString(Path.of("target/test-classes/sqrl/compiled-plan.json"), contents, UTF_8);
+    Files.writeString(
+        Path.of("target/test-classes/savepoint-test/compiled-plan.json"), contents, UTF_8);
   }
 
   interface TransactionDao {

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

@@ -19,16 +19,29 @@
 
 import com.google.common.base.Charsets;
 import com.google.common.io.Resources;
+import java.util.stream.Stream;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 class SqlUtilsTest {
 
+  static Stream<Arguments> testArgs() {
+    return Stream.of(
+        Arguments.of("flink.sql", 12),
+        Arguments.of("test_sql.sql", 6),
+        Arguments.of("test_udf_sql.sql", 6));
+  }
+
   @ParameterizedTest(name = "{0}")
-  @CsvSource({"flink.sql,18", "test_sql.sql,6", "test_udf_sql.sql,6"})
+  @MethodSource("testArgs")
   void givenSource_when_thenSplitCorrectly(String filename, int numberOfStatements)
       throws Exception {
-    var script = Resources.toString(getClass().getResource("/sql/" + filename), Charsets.UTF_8);
+
+    var scriptUrl = getClass().getResource("/sql/" + filename);
+    assertThat(scriptUrl).isNotNull();
+
+    var script = Resources.toString(scriptUrl, Charsets.UTF_8);
     var stmts = SqlUtils.parseStatements(script);
     assertThat(stmts).isNotNull().isNotEmpty().hasSize(numberOfStatements);
   }

flink-sql-runner/src/test/resources/config/config.yaml

@@ -1 +1 @@
-"table.exec.source.idle-timeout": "100 ms"
+table.exec.source.idle-timeout: '100ms'