Merge branch 'main' of github.com:DataSQRL/flink-jar-runner into safe-kafka-connectors

Author: velo

.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'

.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: |

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlBaseJdbcRowConverter.java

@@ -48,7 +48,7 @@ public SqrlBaseJdbcRowConverter(RowType rowType) {
   protected JdbcSerializationConverter wrapIntoNullableExternalConverter(
       JdbcSerializationConverter jdbcSerializationConverter, LogicalType type) {
     if (type.getTypeRoot() == TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
-      int timestampWithTimezone = Types.TIMESTAMP_WITH_TIMEZONE;
+      var timestampWithTimezone = Types.TIMESTAMP_WITH_TIMEZONE;
       return (val, index, statement) -> {
         if (val == null || val.isNullAt(index) || LogicalTypeRoot.NULL.equals(type.getTypeRoot())) {
           statement.setNull(index, timestampWithTimezone);
@@ -62,15 +62,15 @@ protected JdbcSerializationConverter wrapIntoNullableExternalConverter(
 
   @Override
   public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
-    LogicalTypeRoot root = type.getTypeRoot();
+    var root = type.getTypeRoot();
 
     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;
+      var arrayType = (ArrayType) type;
       return createArrayConverter(arrayType);
     } else if (root == LogicalTypeRoot.ROW) {
       return val -> val;
@@ -83,16 +83,15 @@ public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
 
   @Override
   protected JdbcSerializationConverter createExternalConverter(LogicalType type) {
-    switch (type.getTypeRoot()) {
-      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-        final int tsPrecision = ((LocalZonedTimestampType) type).getPrecision();
-        return (val, index, statement) ->
+    return switch (type.getTypeRoot()) {
+      case TIMESTAMP_WITH_LOCAL_TIME_ZONE -> {
+        final var tsPrecision = ((LocalZonedTimestampType) type).getPrecision();
+        yield (val, index, statement) ->
             statement.setTimestamp(index, val.getTimestamp(index, tsPrecision).toTimestamp());
-      case MULTISET:
-      case RAW:
-      default:
-        return super.createExternalConverter(type);
-    }
+      }
+      case MULTISET, RAW -> super.createExternalConverter(type);
+      default -> super.createExternalConverter(type);
+    };
   }
 
   @SneakyThrows
@@ -101,10 +100,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());
       }

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlJdbcDynamicTableFactory.java

@@ -71,15 +71,14 @@ public class SqrlJdbcDynamicTableFactory implements DynamicTableSinkFactory {
 
   @Override
   public DynamicTableSink createDynamicTableSink(Context context) {
-    final FactoryUtil.TableFactoryHelper helper =
-        FactoryUtil.createTableFactoryHelper(this, context);
-    final ReadableConfig config = helper.getOptions();
+    final var helper = FactoryUtil.createTableFactoryHelper(this, context);
+    final var config = helper.getOptions();
 
     helper.validate();
     validateConfigOptions(config, context.getClassLoader());
     validateDataTypeWithJdbcDialect(
         context.getPhysicalRowDataType(), config.get(URL), context.getClassLoader());
-    InternalJdbcConnectionOptions jdbcOptions = getJdbcOptions(config, context.getClassLoader());
+    var jdbcOptions = getJdbcOptions(config, context.getClassLoader());
 
     return new JdbcDynamicTableSink(
         jdbcOptions,
@@ -91,15 +90,15 @@ public DynamicTableSink createDynamicTableSink(Context context) {
 
   private static void validateDataTypeWithJdbcDialect(
       DataType dataType, String url, ClassLoader classLoader) {
-    JdbcDialect dialect = loadDialect(url, classLoader);
+    var dialect = loadDialect(url, classLoader);
 
     dialect.validate((RowType) dataType.getLogicalType());
   }
 
   private InternalJdbcConnectionOptions getJdbcOptions(
       ReadableConfig readableConfig, ClassLoader classLoader) {
-    final String url = readableConfig.get(URL);
-    final InternalJdbcConnectionOptions.Builder builder =
+    final var url = readableConfig.get(URL);
+    final var builder =
         InternalJdbcConnectionOptions.builder()
             .setClassLoader(classLoader)
             .setDBUrl(url)
@@ -116,7 +115,7 @@ private InternalJdbcConnectionOptions getJdbcOptions(
   }
 
   private static JdbcDialect loadDialect(String url, ClassLoader classLoader) {
-    JdbcDialect dialect = JdbcDialectLoader.load(url, classLoader);
+    var dialect = JdbcDialectLoader.load(url, classLoader);
     // sqrl: standard postgres dialect with extended dialect
     if (dialect.dialectName().equalsIgnoreCase("PostgreSQL")) {
       return new SqrlPostgresDialect();
@@ -125,7 +124,7 @@ private static JdbcDialect loadDialect(String url, ClassLoader classLoader) {
   }
 
   private JdbcExecutionOptions getJdbcExecutionOptions(ReadableConfig config) {
-    final JdbcExecutionOptions.Builder builder = new JdbcExecutionOptions.Builder();
+    final var builder = new JdbcExecutionOptions.Builder();
     builder.withBatchSize(config.get(SINK_BUFFER_FLUSH_MAX_ROWS));
     builder.withBatchIntervalMs(config.get(SINK_BUFFER_FLUSH_INTERVAL).toMillis());
     builder.withMaxRetries(config.get(SINK_MAX_RETRIES));
@@ -135,7 +134,7 @@ private JdbcExecutionOptions getJdbcExecutionOptions(ReadableConfig config) {
   private JdbcDmlOptions getJdbcDmlOptions(
       InternalJdbcConnectionOptions jdbcOptions, DataType dataType, int[] primaryKeyIndexes) {
 
-    String[] keyFields =
+    var keyFields =
         Arrays.stream(primaryKeyIndexes)
             .mapToObj(i -> DataType.getFieldNames(dataType).get(i))
             .toArray(String[]::new);
@@ -209,7 +208,7 @@ public Set<ConfigOption<?>> forwardOptions() {
   }
 
   private void validateConfigOptions(ReadableConfig config, ClassLoader classLoader) {
-    String jdbcUrl = config.get(URL);
+    var jdbcUrl = config.get(URL);
     //        JdbcDialectLoader.load(jdbcUrl, classLoader);
 
     checkAllOrNone(config, new ConfigOption[] {USERNAME, PASSWORD});
@@ -229,50 +228,47 @@ 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())));
     }
   }
 
   private void checkAllOrNone(ReadableConfig config, ConfigOption<?>[] configOptions) {
-    int presentCount = 0;
+    var presentCount = 0;
     for (ConfigOption configOption : configOptions) {
       if (config.getOptional(configOption).isPresent()) {
         presentCount++;
       }
     }
-    String[] propertyNames =
-        Arrays.stream(configOptions).map(ConfigOption::key).toArray(String[]::new);
+    var propertyNames = Arrays.stream(configOptions).map(ConfigOption::key).toArray(String[]::new);
     Preconditions.checkArgument(
         configOptions.length == presentCount || presentCount == 0,
         "Either all or none of the following options should be provided:\n"

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
@@ -67,9 +68,9 @@ public Optional<String> defaultDriverName() {
   @Override
   public Optional<String> getUpsertStatement(
       String tableName, String[] fieldNames, String[] uniqueKeyFields) {
-    String uniqueColumns =
+    var uniqueColumns =
         Arrays.stream(uniqueKeyFields).map(this::quoteIdentifier).collect(Collectors.joining(", "));
-    String updateClause =
+    var updateClause =
         Arrays.stream(fieldNames)
             .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
             .collect(Collectors.joining(", "));
@@ -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);

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>>
@@ -98,21 +99,18 @@ public JdbcDeserializationConverter createArrayConverter(ArrayType arrayType) {
     // primitive byte arrays
     final Class<?> elementClass =
         LogicalTypeUtils.toInternalConversionClass(arrayType.getElementType());
-    final JdbcDeserializationConverter elementConverter =
-        createNullableInternalConverter(arrayType.getElementType());
+    final var elementConverter = createNullableInternalConverter(arrayType.getElementType());
     return val -> {
       // 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;
       }
-      final Object[] array =
-          (Object[]) java.lang.reflect.Array.newInstance(elementClass, in.length);
-      for (int i = 0; i < in.length; i++) {
+      final var array = (Object[]) java.lang.reflect.Array.newInstance(elementClass, in.length);
+      for (var i = 0; i < in.length; i++) {
         array[i] = elementConverter.deserialize(in[i]);
       }
       return new GenericArrayData(array);

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/FlinkArrayTypeUtil.java

@@ -21,42 +21,41 @@
 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) {
+      var elementType = arrayType.getElementType();
       return isScalar(elementType) || isScalarArray(elementType);
     }
     return false;
   }
 
   public static boolean isScalar(LogicalType type) {
-    switch (type.getTypeRoot()) {
-      case BOOLEAN:
-      case TINYINT:
-      case SMALLINT:
-      case INTEGER:
-      case BIGINT:
-      case FLOAT:
-      case DOUBLE:
-      case CHAR:
-      case VARCHAR:
-      case BINARY:
-      case VARBINARY:
-      case DATE:
-      case TIME_WITHOUT_TIME_ZONE:
-      case TIMESTAMP_WITH_TIME_ZONE:
-      case TIMESTAMP_WITHOUT_TIME_ZONE:
-      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-      case DECIMAL:
-        return true;
-      default:
-        return false;
-    }
+    return switch (type.getTypeRoot()) {
+      case BOOLEAN,
+              TINYINT,
+              SMALLINT,
+              INTEGER,
+              BIGINT,
+              FLOAT,
+              DOUBLE,
+              CHAR,
+              VARCHAR,
+              BINARY,
+              VARBINARY,
+              DATE,
+              TIME_WITHOUT_TIME_ZONE,
+              TIMESTAMP_WITH_TIME_ZONE,
+              TIMESTAMP_WITHOUT_TIME_ZONE,
+              TIMESTAMP_WITH_LOCAL_TIME_ZONE,
+              DECIMAL ->
+          true;
+      default -> false;
+    };
   }
 }