Applied rewrite java 17 migration

velo · velo · commit ae9b5918935c · 2025-04-21T20:24:53.000-03:00
$ mvn clean -U org.openrewrite.maven:rewrite-maven-plugin:run -Drewrite.recipeArtifactCoordinates=org.openrewrite.recipe:rewrite-migrate-java:RELEASE -Drewrite.activeRecipes=org.openrewrite.java.migrate.UpgradeToJava17 -Drewrite.exportDatatables=true

Signed-off-by: Marvin Froeder &lt;marvin@datasqrl.com&gt;
diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml
@@ -32,7 +32,7 @@ jobs:
         uses: actions/setup-java@v3
         with:
           distribution: 'temurin'
-          java-version: '11'
+          java-version: '17'
 
       - name: Import GPG Key
         if: github.event_name != 'pull_request'
diff --git a/.github/workflows/uber-jar.yml b/.github/workflows/uber-jar.yml
@@ -36,7 +36,7 @@ jobs:
         uses: actions/setup-java@v3
         with:
           distribution: 'temurin'
-          java-version: '11'
+          java-version: '17'
 
       - name: Generate settings.xml
         run: |
diff --git a/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlBaseJdbcRowConverter.java b/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlBaseJdbcRowConverter.java
@@ -66,8 +66,8 @@ public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
 
     if (root == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
       return val ->
-          val instanceof LocalDateTime
-              ? TimestampData.fromLocalDateTime((LocalDateTime) val)
+          val instanceof LocalDateTime ldt
+              ? TimestampData.fromLocalDateTime(ldt)
               : TimestampData.fromTimestamp((Timestamp) val);
     } else if (root == LogicalTypeRoot.ARRAY) {
       ArrayType arrayType = (ArrayType) type;
@@ -101,10 +101,10 @@ private void createSqlArrayObject(
     // Scalar arrays of any dimension are one array call
     if (isScalarArray(type)) {
       Object[] boxed;
-      if (data instanceof GenericArrayData) {
-        boxed = ((GenericArrayData) data).toObjectArray();
-      } else if (data instanceof BinaryArrayData) {
-        boxed = ((BinaryArrayData) data).toObjectArray(getBaseFlinkArrayType(type));
+      if (data instanceof GenericArrayData arrayData) {
+        boxed = arrayData.toObjectArray();
+      } else if (data instanceof BinaryArrayData arrayData) {
+        boxed = arrayData.toObjectArray(getBaseFlinkArrayType(type));
       } else {
         throw new RuntimeException("Unsupported ArrayData type: " + data.getClass());
       }
diff --git a/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlJdbcDynamicTableFactory.java b/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlJdbcDynamicTableFactory.java
@@ -229,38 +229,36 @@ private void validateConfigOptions(ReadableConfig config, ClassLoader classLoade
       long upperBound = config.get(SCAN_PARTITION_UPPER_BOUND);
       if (lowerBound > upperBound) {
         throw new IllegalArgumentException(
-            String.format(
-                "'%s'='%s' must not be larger than '%s'='%s'.",
-                SCAN_PARTITION_LOWER_BOUND.key(),
-                lowerBound,
-                SCAN_PARTITION_UPPER_BOUND.key(),
-                upperBound));
+            "'%s'='%s' must not be larger than '%s'='%s'."
+                .formatted(
+                    SCAN_PARTITION_LOWER_BOUND.key(),
+                    lowerBound,
+                    SCAN_PARTITION_UPPER_BOUND.key(),
+                    upperBound));
       }
     }
 
     checkAllOrNone(config, new ConfigOption[] {LOOKUP_CACHE_MAX_ROWS, LOOKUP_CACHE_TTL});
 
     if (config.get(LOOKUP_MAX_RETRIES) < 0) {
       throw new IllegalArgumentException(
-          String.format(
-              "The value of '%s' option shouldn't be negative, but is %s.",
-              LOOKUP_MAX_RETRIES.key(), config.get(LOOKUP_MAX_RETRIES)));
+          "The value of '%s' option shouldn't be negative, but is %s."
+              .formatted(LOOKUP_MAX_RETRIES.key(), config.get(LOOKUP_MAX_RETRIES)));
     }
 
     if (config.get(SINK_MAX_RETRIES) < 0) {
       throw new IllegalArgumentException(
-          String.format(
-              "The value of '%s' option shouldn't be negative, but is %s.",
-              SINK_MAX_RETRIES.key(), config.get(SINK_MAX_RETRIES)));
+          "The value of '%s' option shouldn't be negative, but is %s."
+              .formatted(SINK_MAX_RETRIES.key(), config.get(SINK_MAX_RETRIES)));
     }
 
     if (config.get(MAX_RETRY_TIMEOUT).getSeconds() <= 0) {
       throw new IllegalArgumentException(
-          String.format(
-              "The value of '%s' option must be in second granularity and shouldn't be smaller than 1 second, but is %s.",
-              MAX_RETRY_TIMEOUT.key(),
-              config.get(
-                  ConfigOptions.key(MAX_RETRY_TIMEOUT.key()).stringType().noDefaultValue())));
+          "The value of '%s' option must be in second granularity and shouldn't be smaller than 1 second, but is %s."
+              .formatted(
+                  MAX_RETRY_TIMEOUT.key(),
+                  config.get(
+                      ConfigOptions.key(MAX_RETRY_TIMEOUT.key()).stringType().noDefaultValue())));
     }
   }
 
diff --git a/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresDialect.java b/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresDialect.java
@@ -15,6 +15,7 @@
  */
 package com.datasqrl.connector.postgresql.jdbc;
 
+import java.io.Serial;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
@@ -36,7 +37,7 @@
  */
 public class SqrlPostgresDialect extends AbstractDialect {
 
-  private static final long serialVersionUID = 1L;
+  @Serial private static final long serialVersionUID = 1L;
 
   // Define MAX/MIN precision of TIMESTAMP type according to PostgreSQL docs:
   // https://www.postgresql.org/docs/12/datatype-datetime.html
@@ -93,8 +94,8 @@ public void validate(RowType rowType) throws ValidationException {
 
     if (!unsupportedTypes.isEmpty()) {
       throw new ValidationException(
-          String.format(
-              "The %s dialect doesn't support type: %s.", this.dialectName(), unsupportedTypes));
+          "The %s dialect doesn't support type: %s."
+              .formatted(this.dialectName(), unsupportedTypes));
     }
 
     super.validate(rowType);
diff --git a/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresRowConverter.java b/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresRowConverter.java
@@ -16,6 +16,7 @@
 package com.datasqrl.connector.postgresql.jdbc;
 
 import com.datasqrl.connector.postgresql.type.JdbcTypeSerializer;
+import java.io.Serial;
 import java.lang.reflect.Type;
 import java.util.Map;
 import java.util.ServiceLoader;
@@ -35,7 +36,7 @@
  */
 public class SqrlPostgresRowConverter extends SqrlBaseJdbcRowConverter {
 
-  private static final long serialVersionUID = 1L;
+  @Serial private static final long serialVersionUID = 1L;
 
   public static final Map<
           Type, JdbcTypeSerializer<JdbcDeserializationConverter, JdbcSerializationConverter>>
@@ -104,8 +105,7 @@ public JdbcDeserializationConverter createArrayConverter(ArrayType arrayType) {
       // sqrl: check if scalar array
 
       Object[] in;
-      if (val instanceof PgArray) {
-        PgArray pgArray = (PgArray) val;
+      if (val instanceof PgArray pgArray) {
         in = (Object[]) pgArray.getArray();
       } else {
         in = (Object[]) val;
diff --git a/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/FlinkArrayTypeUtil.java b/connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/FlinkArrayTypeUtil.java
@@ -21,15 +21,15 @@
 public class FlinkArrayTypeUtil {
 
   public static LogicalType getBaseFlinkArrayType(LogicalType type) {
-    if (type instanceof ArrayType) {
-      return getBaseFlinkArrayType(((ArrayType) type).getElementType());
+    if (type instanceof ArrayType arrayType) {
+      return getBaseFlinkArrayType(arrayType.getElementType());
     }
     return type;
   }
 
   public static boolean isScalarArray(LogicalType type) {
-    if (type instanceof ArrayType) {
-      LogicalType elementType = ((ArrayType) type).getElementType();
+    if (type instanceof ArrayType arrayType) {
+      LogicalType elementType = arrayType.getElementType();
       return isScalar(elementType) || isScalarArray(elementType);
     }
     return false;
diff --git a/flink-sql-runner/pom.xml b/flink-sql-runner/pom.xml
@@ -476,16 +476,7 @@
         <flink.version>1.19.2</flink.version>
         <jdbc.version>3.2.0-1.19</jdbc.version>
         <kafka.version>3.2.0-1.19</kafka.version>
-        <flink-base-image>1.19.2-scala_2.12-java11</flink-base-image>
-      </properties>
-    </profile>
-    <profile>
-      <id>flink-1.18</id>
-      <properties>
-        <flink.version>1.18.1</flink.version>
-        <jdbc.version>3.2.0-1.18</jdbc.version>
-        <kafka.version>3.2.0-1.18</kafka.version>
-        <flink-base-image>1.18.1-scala_2.12-java11</flink-base-image>
+        <flink-base-image>1.19.2-scala_2.12-java17</flink-base-image>
       </properties>
     </profile>
 
diff --git a/flink-sql-runner/src/main/java/com/datasqrl/EnvironmentVariablesUtils.java b/flink-sql-runner/src/main/java/com/datasqrl/EnvironmentVariablesUtils.java
@@ -43,7 +43,7 @@ public static String replaceWithEnv(String command, Map<String, String> envVaria
       String key = matcher.group(1);
       String envValue = envVariables.get(key);
       if (envValue == null) {
-        throw new IllegalStateException(String.format("Missing environment variable: %s", key));
+        throw new IllegalStateException("Missing environment variable: %s".formatted(key));
       }
       matcher.appendReplacement(result, Matcher.quoteReplacement(envValue));
     }
diff --git a/flink-sql-runner/src/main/java/com/datasqrl/FlinkMain.java b/flink-sql-runner/src/main/java/com/datasqrl/FlinkMain.java
@@ -111,9 +111,8 @@ public int run() throws Exception {
           EnvironmentVariablesUtils.validateEnvironmentVariables(script);
       if (!missingEnvironmentVariables.isEmpty()) {
         throw new IllegalStateException(
-            String.format(
-                "Could not find the following environment variables: %s",
-                missingEnvironmentVariables));
+            "Could not find the following environment variables: %s"
+                .formatted(missingEnvironmentVariables));
       }
 
       tableResult = sqlExecutor.executeScript(script);
@@ -125,9 +124,8 @@ public int run() throws Exception {
           EnvironmentVariablesUtils.validateEnvironmentVariables(planJson);
       if (!missingEnvironmentVariables.isEmpty()) {
         throw new IllegalStateException(
-            String.format(
-                "Could not find the following environment variables: %s",
-                missingEnvironmentVariables));
+            "Could not find the following environment variables: %s"
+                .formatted(missingEnvironmentVariables));
       }
 
       planJson = replaceScriptWithEnv(planJson);
diff --git a/flink-sql-runner/src/main/java/com/datasqrl/SqlExecutor.java b/flink-sql-runner/src/main/java/com/datasqrl/SqlExecutor.java
@@ -66,9 +66,10 @@ public void setupSystemFunctions() {
       standardLibraryFunctions.forEach(
           function -> {
             String sql =
-                String.format(
-                    "CREATE TEMPORARY FUNCTION IF NOT EXISTS `%s` AS '%s' LANGUAGE JAVA;",
-                    getFunctionNameFromClass(function.getClass()), function.getClass().getName());
+                "CREATE TEMPORARY FUNCTION IF NOT EXISTS `%s` AS '%s' LANGUAGE JAVA;"
+                    .formatted(
+                        getFunctionNameFromClass(function.getClass()),
+                        function.getClass().getName());
 
             System.out.println(sql);
 
diff --git a/flink-sql-runner/src/test/java/com/datasqrl/CommandLineUtil.java b/flink-sql-runner/src/test/java/com/datasqrl/CommandLineUtil.java
@@ -65,8 +65,7 @@ public static String execute(Path workDir, String command) throws ExecuteExcepti
     } 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;
+      if (e instanceof ExecuteException ee) {
         throw new ExecuteException(result, ee.getExitValue(), ee);
       }
       throw new RuntimeException(e);
diff --git a/formats/flexible-json-format/src/main/java/com/datasqrl/flink/format/json/SqrlJsonRowDataSerializationSchema.java b/formats/flexible-json-format/src/main/java/com/datasqrl/flink/format/json/SqrlJsonRowDataSerializationSchema.java
@@ -31,6 +31,7 @@
                                          * limitations under the License.
                                          */
 
+import java.io.Serial;
 import java.util.Objects;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -59,7 +60,7 @@
  */
 @Internal
 public class SqrlJsonRowDataSerializationSchema implements SerializationSchema<RowData> {
-  private static final long serialVersionUID = 1L;
+  @Serial private static final long serialVersionUID = 1L;
 
   /** RowType to generate the runtime converter. */
   private final RowType rowType;
@@ -118,7 +119,7 @@ public byte[] serialize(RowData row) {
       runtimeConverter.convert(mapper, node, row);
       return mapper.writeValueAsBytes(node);
     } catch (Throwable t) {
-      throw new RuntimeException(String.format("Could not serialize row '%s'.", row), t);
+      throw new RuntimeException("Could not serialize row '%s'.".formatted(row), t);
     }
   }
 
diff --git a/pom.xml b/pom.xml
@@ -98,8 +98,8 @@
     <lombok.version>1.18.38</lombok.version>
 
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <maven.compiler.source>11</maven.compiler.source>
-    <maven.compiler.target>11</maven.compiler.target>
+    <maven.compiler.source>17</maven.compiler.source>
+    <maven.compiler.target>17</maven.compiler.target>
 
     <!-- Example: skip install hooks, etc. -->
     <gcf.skipInstallHooks>true</gcf.skipInstallHooks>
diff --git a/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonArray.java b/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonArray.java
@@ -40,8 +40,7 @@ public FlinkJsonType eval(Object... objects) {
     ArrayNode arrayNode = mapper.createArrayNode();
 
     for (Object value : objects) {
-      if (value instanceof FlinkJsonType) {
-        FlinkJsonType type = (FlinkJsonType) value;
+      if (value instanceof FlinkJsonType type) {
         arrayNode.add(type.json);
       } else {
         arrayNode.addPOJO(value);
diff --git a/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonArrayAgg.java b/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonArrayAgg.java
@@ -91,8 +91,8 @@ public void merge(ArrayAgg accumulator, java.lang.Iterable<ArrayAgg> iterable) {
   public FlinkJsonType getValue(ArrayAgg accumulator) {
     ArrayNode arrayNode = mapper.createArrayNode();
     for (Object o : accumulator.getObjects()) {
-      if (o instanceof FlinkJsonType) {
-        arrayNode.add(((FlinkJsonType) o).json);
+      if (o instanceof FlinkJsonType type) {
+        arrayNode.add(type.json);
       } else {
         arrayNode.addPOJO(o);
       }
diff --git a/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonObject.java b/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonObject.java
@@ -54,8 +54,7 @@ public FlinkJsonType eval(Object... objects) {
       }
       String key = (String) objects[i];
       Object value = objects[i + 1];
-      if (value instanceof FlinkJsonType) {
-        FlinkJsonType type = (FlinkJsonType) value;
+      if (value instanceof FlinkJsonType type) {
         objectNode.put(key, type.json);
       } else {
         objectNode.putPOJO(key, value);
diff --git a/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonObjectAgg.java b/types/json-type/src/main/java/com/datasqrl/types/json/functions/JsonObjectAgg.java
@@ -51,8 +51,8 @@ public void accumulate(ObjectAgg accumulator, String key, String value) {
 
   public void accumulate(
       ObjectAgg accumulator, String key, @DataTypeHint(inputGroup = InputGroup.ANY) Object value) {
-    if (value instanceof FlinkJsonType) {
-      accumulateObject(accumulator, key, ((FlinkJsonType) value).getJson());
+    if (value instanceof FlinkJsonType type) {
+      accumulateObject(accumulator, key, type.getJson());
     } else {
       accumulator.add(key, mapper.getNodeFactory().pojoNode(value));
     }
diff --git a/types/json-type/src/main/java/com/datasqrl/types/json/functions/ToJson.java b/types/json-type/src/main/java/com/datasqrl/types/json/functions/ToJson.java
@@ -50,16 +50,15 @@ public FlinkJsonType eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object json
     if (json == null) {
       return null;
     }
-    if (json instanceof FlinkJsonType) {
-      return (FlinkJsonType) json;
+    if (json instanceof FlinkJsonType type) {
+      return type;
     }
 
     return new FlinkJsonType(unboxFlinkToJsonNode(json));
   }
 
   JsonNode unboxFlinkToJsonNode(Object json) {
-    if (json instanceof Row) {
-      Row row = (Row) json;
+    if (json instanceof Row row) {
       ObjectNode objectNode = mapper.createObjectNode();
       String[] fieldNames =
           row.getFieldNames(true).toArray(new String[0]); // Get field names in an array
@@ -68,8 +67,7 @@ JsonNode unboxFlinkToJsonNode(Object json) {
         objectNode.set(fieldName, unboxFlinkToJsonNode(field)); // Recursively unbox each field
       }
       return objectNode;
-    } else if (json instanceof Row[]) {
-      Row[] rows = (Row[]) json;
+    } else if (json instanceof Row[] rows) {
       ArrayNode arrayNode = mapper.createArrayNode();
       for (Row row : rows) {
         if (row == null) {

Original file line number	Diff line number	Diff line change
`@@ -21,15 +21,15 @@`
`21`	`21`	`public class FlinkArrayTypeUtil {`
`22`	`22`
`23`	`23`	`public static LogicalType getBaseFlinkArrayType(LogicalType type) {`
`24`		`- if (type instanceof ArrayType) {`
`25`		`- return getBaseFlinkArrayType(((ArrayType) type).getElementType());`
	`24`	`+ if (type instanceof ArrayType arrayType) {`
	`25`	`+ return getBaseFlinkArrayType(arrayType.getElementType());`
`26`	`26`	`}`
`27`	`27`	`return type;`
`28`	`28`	`}`
`29`	`29`
`30`	`30`	`public static boolean isScalarArray(LogicalType type) {`
`31`		`- if (type instanceof ArrayType) {`
`32`		`- LogicalType elementType = ((ArrayType) type).getElementType();`
	`31`	`+ if (type instanceof ArrayType arrayType) {`
	`32`	`+ LogicalType elementType = arrayType.getElementType();`
`33`	`33`	`return isScalar(elementType) \|\| isScalarArray(elementType);`
`34`	`34`	`}`
`35`	`35`	`return false;`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public static String replaceWithEnv(String command, Map<String, String> envVaria`
`43`	`43`	`String key = matcher.group(1);`
`44`	`44`	`String envValue = envVariables.get(key);`
`45`	`45`	`if (envValue == null) {`
`46`		`- throw new IllegalStateException(String.format("Missing environment variable: %s", key));`
	`46`	`+ throw new IllegalStateException("Missing environment variable: %s".formatted(key));`
`47`	`47`	`}`
`48`	`48`	`matcher.appendReplacement(result, Matcher.quoteReplacement(envValue));`
`49`	`49`	`}`