Bump Flink to latest patch versions, fix logging (#169)

Author: ferenc-csaky

.github/workflows/uber-jar.yml

@@ -55,7 +55,6 @@ jobs:
         run: |
           mvn -B org.apache.maven.plugins:maven-dependency-plugin:3.8.1:go-offline de.qaware.maven:go-offline-maven-plugin:1.2.8:resolve-dependencies -P${{ matrix.FLINK_PROFILE }}
           mvn -B -f flink-sql-runner/pom.xml org.apache.maven.plugins:maven-resources-plugin:3.3.1:resources -P${{ matrix.FLINK_PROFILE }}
-          docker pull postgres:17
           grep '^FROM' flink-sql-runner/target/Dockerfile | awk '{print $2}' | xargs -n1 docker pull
 
       - name: Get project version and store in env

connectors/postgresql-connector/src/test/java/com/datasqrl/flinkrunner/connector/postgresql/FlinkJdbcTest.java

@@ -21,6 +21,7 @@
 import com.datasqrl.flinkrunner.stdlib.json.to_jsonb;
 import java.io.IOException;
 import java.sql.DriverManager;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.core.memory.DataOutputSerializer;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -34,23 +35,26 @@
 import org.testcontainers.containers.PostgreSQLContainer;
 
 @ExtendWith(MiniClusterExtension.class)
+@Slf4j
 public class FlinkJdbcTest {
 
   public static void main(String[] args) throws IOException {
     var input =
         new DataInputDeserializer(
             EncodingUtils.decodeBase64ToBytes(
                 "ADFjb20uZGF0YXNxcmwuanNvbi5GbGlua0pzb25UeXBlU2VyaWFsaXplclNuYXBzaG90AAAAAQApY29tLmRhdGFzcXJsLmpzb24uRmxpbmtKc29uVHlwZVNlcmlhbGl6ZXI="));
-    System.out.println(input.readUTF());
-    System.out.println(input.readInt());
-    System.out.println(input.readUTF());
+
+    log.debug(input.readUTF());
+    log.debug("{}", input.readInt());
+    log.debug(input.readUTF());
 
     var output = new DataOutputSerializer(91);
     var snapshot = new FlinkJsonTypeSerializerSnapshot();
     output.writeUTF(snapshot.getClass().getName());
     output.writeInt(snapshot.getCurrentVersion());
     snapshot.writeSnapshot(output);
-    System.out.println(EncodingUtils.encodeBytesToBase64(output.getSharedBuffer()));
+
+    log.debug(EncodingUtils.encodeBytesToBase64(output.getSharedBuffer()));
   }
 
   @Test

flink-sql-runner/pom.xml

@@ -156,21 +156,18 @@
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>
-      <version>${log4j.version}</version>
       <scope>runtime</scope>
     </dependency>
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-api</artifactId>
-      <version>${log4j.version}</version>
       <scope>runtime</scope>
     </dependency>
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-core</artifactId>
-      <version>${log4j.version}</version>
       <scope>runtime</scope>
     </dependency>
 
@@ -216,6 +213,12 @@
       <artifactId>com.nextbreakpoint.flink.client</artifactId>
       <version>${nextbreakpoint.flink.client.version}</version>
       <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.logging.log4j</groupId>
+          <artifactId>log4j-slf4j2-impl</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
 
     <dependency>
@@ -385,7 +388,7 @@
             <configuration>
               <workingDirectory>${project.basedir}/target/</workingDirectory>
               <executable>docker</executable>
-              <skip>${skipTests}</skip>
+              <skip>${fast}</skip>
               <arguments>
                 <argument>build</argument>
                 <argument>--tag</argument>

flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java

@@ -20,6 +20,7 @@
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.RuntimeExecutionMode;
@@ -30,6 +31,7 @@
 import org.apache.flink.util.FileUtils;
 
 @RequiredArgsConstructor
+@Slf4j
 abstract class BaseRunner {
 
   final RuntimeExecutionMode mode;
@@ -88,7 +90,7 @@ TableResult run(Supplier<SqlExecutor> sqlExecutorSupplier) throws Exception {
   Configuration initConfiguration() {
     var conf = new Configuration();
     if (StringUtils.isNotBlank(configDir)) {
-      System.out.printf("Loading configuration from %s\n", configDir);
+      log.info("Loading Flink configuration from '{}'", configDir);
       conf = GlobalConfiguration.loadConfiguration(configDir);
     }
 

flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/CliRunner.java

@@ -18,12 +18,14 @@
 import java.util.Arrays;
 import java.util.concurrent.Callable;
 import javax.annotation.Nullable;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.RuntimeExecutionMode;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 
+@Slf4j
 public class CliRunner extends BaseRunner {
 
   @SuppressWarnings("unused")
@@ -88,7 +90,7 @@ public CliRunner(
   }
 
   public static void main(String[] args) throws Exception {
-    System.out.printf("\n\nExecuting flink-sql-runner: %s\n\n", Arrays.toString(args));
+    log.info("Executing flink-sql-runner: {}", Arrays.toString(args));
 
     var cl = new CommandLine(new SqlRunner());
     var resCode = cl.execute(args);
@@ -110,6 +112,6 @@ public static void main(String[] args) throws Exception {
     new CliRunner(runner.mode, runner.sqlFile, runner.planFile, runner.configDir, runner.udfPath)
         .run();
 
-    System.out.println("Finished flink-sql-runner");
+    log.info("Finished flink-sql-runner execution");
   }
 }

flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/SqlExecutor.java

@@ -188,7 +188,6 @@ private void setupUdfPath(String udfPath) {
       }
     } catch (Exception e) {
       log.error("Failed to set up UDF path", e);
-      e.printStackTrace(System.out);
     }
   }
 

flink-sql-runner/src/main/resources/log4j2.properties

@@ -16,17 +16,15 @@
 
 # Set root logger level
 rootLogger.level = INFO
+rootLogger.appenderRef.console.ref = MainAppender
+
+# Set logging level for org.apache.flink to WARN
+logger.flink.name = org.apache.flink
+logger.flink.level = WARN
 
 # Define Console appender
+appender.console.name = MainAppender
 appender.console.type = Console
-appender.console.name = Console
 appender.console.target = SYSTEM_OUT
 appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n
-
-# Assign appender to root logger
-rootLogger.appenderRef.console.ref = Console
-
-# Set logging level for org.apache.flink to WARN
-logger.flink.name = org.apache.flink
-logger.flink.level = WARN

flink-sql-runner/src/main/resources/log4j2.xml

@@ -28,6 +28,7 @@
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
 import org.jdbi.v3.core.statement.SqlLogger;
 import org.jdbi.v3.core.statement.StatementContext;
@@ -39,6 +40,7 @@
 import org.junit.jupiter.api.io.TempDir;
 import org.testcontainers.containers.Container;
 
+@Slf4j
 public class SavepointIT extends AbstractITSupport {
 
   @TempDir private Path tempDir;
@@ -72,8 +74,8 @@ void givenPlanScript_whenSavepoint_thenResumeFromSavepointSuccessfully() throws
         flinkContainer.execInContainer(
             "flink", "stop", jobId, "--savepointPath", CONTAINER_TEST_OUT_PATH);
 
-    System.out.printf("Flink container STDOUT: %s%n", cmdRes.getStdout());
-    System.out.printf("Flink container STDERR: %s%n", cmdRes.getStderr());
+    log.debug("Flink container STDOUT: {}", cmdRes.getStdout());
+    log.debug("Flink container STDERR: {}", cmdRes.getStderr());
 
     // check if STOPed, make sure no new records are create
     untilAssert(
@@ -126,17 +128,19 @@ private TransactionDao connect() {
     jdbi.setSqlLogger(
         new SqlLogger() {
           @Override
-          public void logAfterExecution(StatementContext context) {
-            System.out.printf(
-                "Executed in '%s' with parameters '%s'\n",
-                context.getParsedSql().getSql(), context.getBinding());
+          public void logAfterExecution(StatementContext ctx) {
+            log.info(
+                "Executed '{}' with parameters '{}'",
+                ctx.getParsedSql().getSql(),
+                ctx.getBinding());
           }
 
           @Override
-          public void logException(StatementContext context, SQLException ex) {
-            System.out.printf(
-                "Exception while executing '%s' with parameters '%s'\n",
-                context.getParsedSql().getSql(), context.getBinding(), ex);
+          public void logException(StatementContext ctx, SQLException ex) {
+            log.info(
+                "Exception while executing '{}' with parameters '{}'",
+                ctx.getParsedSql().getSql(),
+                ctx.getBinding());
           }
         });