CDK module changes for destination (airbytehq#36588)

### TL;DR
Updated JDBC and S3 destination to compile destination-redshift

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/factory/ConnectionFactory.kt

@@ -23,6 +23,7 @@ object ConnectionFactory {
      * @param jdbcConnectionString The JDBC connection string.
      * @return The configured [Connection]
      */
+    @JvmStatic
     fun create(
         username: String?,
         password: String?,

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/factory/DatabaseDriver.kt

@@ -4,10 +4,7 @@
 package io.airbyte.cdk.db.factory
 
 /** Collection of JDBC driver class names and the associated JDBC URL format string. */
-enum class DatabaseDriver(
-    @JvmField val driverClassName: String,
-    @JvmField val urlFormatString: String
-) {
+enum class DatabaseDriver(val driverClassName: String, val urlFormatString: String) {
     CLICKHOUSE("com.clickhouse.jdbc.ClickHouseDriver", "jdbc:clickhouse:%s://%s:%d/%s"),
     DATABRICKS(
         "com.databricks.client.jdbc.Driver",

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.kt

@@ -213,15 +213,20 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putTime(node: ObjectNode, columnName: String?, resultSet: ResultSet, index: Int) {
+    protected open fun putTime(
+        node: ObjectNode,
+        columnName: String?,
+        resultSet: ResultSet,
+        index: Int
+    ) {
         node.put(
             columnName,
             DateTimeConverter.convertToTime(getObject(resultSet, index, LocalTime::class.java))
         )
     }
 
     @Throws(SQLException::class)
-    protected fun putTimestamp(
+    protected open fun putTimestamp(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,
@@ -419,7 +424,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putTimeWithTimezone(
+    protected open fun putTimeWithTimezone(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,
@@ -430,7 +435,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putTimestampWithTimezone(
+    protected open fun putTimestampWithTimezone(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/JdbcSourceOperations.kt

@@ -15,7 +15,7 @@ import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
 /** Implementation of source operations with standard JDBC types. */
-class JdbcSourceOperations :
+open class JdbcSourceOperations :
     AbstractJdbcCompatibleSourceOperations<JDBCType>(), SourceOperations<ResultSet, JDBCType> {
     protected fun safeGetJdbcType(columnTypeInt: Int): JDBCType {
         return try {

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/ssh/SshTunnel.kt

@@ -422,6 +422,7 @@ constructor(
 
         const val TIMEOUT_MILLIS: Int = 15000 // 15 seconds
 
+        @JvmStatic
         fun getInstance(config: JsonNode, hostKey: List<String>, portKey: List<String>): SshTunnel {
             val tunnelMethod =
                 Jsons.getOptional(config, "tunnel_method", "tunnel_method")
@@ -515,6 +516,7 @@ constructor(
             )
         }
 
+        @JvmStatic
         @Throws(Exception::class)
         fun sshWrap(
             config: JsonNode,
@@ -528,6 +530,7 @@ constructor(
             }
         }
 
+        @JvmStatic
         @Throws(Exception::class)
         fun sshWrap(
             config: JsonNode,
@@ -540,6 +543,7 @@ constructor(
             }
         }
 
+        @JvmStatic
         @Throws(Exception::class)
         fun <T> sshWrap(
             config: JsonNode,
@@ -552,6 +556,7 @@ constructor(
             }
         }
 
+        @JvmStatic
         @Throws(Exception::class)
         fun <T> sshWrap(
             config: JsonNode,

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/AbstractJdbcDestination.kt

@@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory
 
 abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationState>(
     driverClass: String,
-    protected val namingResolver: NamingConventionTransformer,
+    protected open val namingResolver: NamingConventionTransformer,
     protected val sqlOperations: SqlOperations
 ) : JdbcConnector(driverClass), Destination {
     protected val configSchemaKey: String
@@ -106,14 +106,14 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
      * @throws Exception
      */
     @Throws(Exception::class)
-    protected fun destinationSpecificTableOperations(database: JdbcDatabase?) {}
+    protected open fun destinationSpecificTableOperations(database: JdbcDatabase?) {}
 
     /**
      * Subclasses which need to modify the DataSource should override [.modifyDataSourceBuilder]
      * rather than this method.
      */
     @VisibleForTesting
-    fun getDataSource(config: JsonNode): DataSource {
+    open fun getDataSource(config: JsonNode): DataSource {
         val jdbcConfig = toJdbcConfig(config)
         val connectionProperties = getConnectionProperties(config)
         val builder =
@@ -137,7 +137,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
     }
 
     @VisibleForTesting
-    fun getDatabase(dataSource: DataSource): JdbcDatabase {
+    open fun getDatabase(dataSource: DataSource): JdbcDatabase {
         return DefaultJdbcDatabase(dataSource)
     }
 
@@ -195,7 +195,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
         return config[JdbcUtils.DATABASE_KEY].asText()
     }
 
-    protected fun getDataTransformer(
+    protected open fun getDataTransformer(
         parsedCatalog: ParsedCatalog?,
         defaultNamespace: String?
     ): StreamAwareDataTransformer {
@@ -343,6 +343,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
          * - set true if need to make attempt to insert dummy records to newly created table. Set
          * false to skip insert step.
          */
+        @JvmStatic
         @Throws(Exception::class)
         fun attemptTableOperations(
             outputSchema: String?,

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/JdbcSqlOperations.kt

@@ -86,7 +86,7 @@ abstract class JdbcSqlOperations : SqlOperations {
         return listOf()
     }
 
-    protected fun createTableQueryV1(schemaName: String?, tableName: String?): String {
+    protected open fun createTableQueryV1(schemaName: String?, tableName: String?): String {
         return String.format(
             """
         CREATE TABLE IF NOT EXISTS %s.%s (
@@ -104,7 +104,7 @@ abstract class JdbcSqlOperations : SqlOperations {
         )
     }
 
-    protected fun createTableQueryV2(schemaName: String?, tableName: String?): String {
+    protected open fun createTableQueryV2(schemaName: String?, tableName: String?): String {
         // Note that Meta is the last column in order, there was a time when tables didn't have
         // meta,
         // we issued Alter to add that column so it should be the last column.

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/SqlOperationsUtils.kt

@@ -29,6 +29,7 @@ object SqlOperationsUtils {
      * @param records records to write
      * @throws SQLException exception
      */
+    @JvmStatic
     @Throws(SQLException::class)
     fun insertRawRecordsInSingleQuery(
         insertQueryComponent: String?,

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/SwitchingDestination.kt

@@ -28,7 +28,7 @@ import org.slf4j.LoggerFactory
  * This class exists to make it easy to define a destination in terms of multiple other destination
  * implementations, switching between them based on the config provided.
  */
-class SwitchingDestination<T : Enum<T>>(
+open class SwitchingDestination<T : Enum<T>>(
     enumClass: Class<T>,
     configToType: Function<JsonNode, T>,
     typeToDestination: Map<T, Destination>

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcDestinationHandler.kt

@@ -263,24 +263,21 @@ abstract class JdbcDestinationHandler<DestinationState>(
                         // We can't directly call record.set here, because that will raise a
                         // ConcurrentModificationException on the fieldnames iterator.
                         // Instead, build up a map of new fields and set them all at once.
-                        newFields.put(fieldName.lowercase(Locale.getDefault()), record[fieldName])
+                        newFields[fieldName.lowercase(Locale.getDefault())] = record[fieldName]
                     }
 
                     record.setAll<JsonNode>(newFields)
                 }
                 .collect(
                     toMap(
                         { record ->
-                            val nameNode: JsonNode = record.get(DESTINATION_STATE_TABLE_COLUMN_NAME)
-                            val namespaceNode: JsonNode =
-                                record.get(DESTINATION_STATE_TABLE_COLUMN_NAMESPACE)
                             AirbyteStreamNameNamespacePair(
-                                if (nameNode != null) nameNode.asText() else null,
-                                if (namespaceNode != null) namespaceNode.asText() else null
+                                record.get(DESTINATION_STATE_TABLE_COLUMN_NAME)?.asText(),
+                                record.get(DESTINATION_STATE_TABLE_COLUMN_NAMESPACE)?.asText()
                             )
                         },
                         { record ->
-                            val stateNode: JsonNode =
+                            val stateNode: JsonNode? =
                                 record.get(DESTINATION_STATE_TABLE_COLUMN_STATE)
                             val state =
                                 if (stateNode != null) Jsons.deserialize(stateNode.asText())

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcSqlGenerator.kt

@@ -144,7 +144,7 @@ abstract class JdbcSqlGenerator(protected val namingTransformer: NamingConventio
         cursorField: Optional<ColumnId>
     ): Field<Int>
 
-    protected val dslContext: DSLContext
+    protected open val dslContext: DSLContext
         get() = DSL.using(dialect)
 
     /**
@@ -596,7 +596,7 @@ abstract class JdbcSqlGenerator(protected val namingTransformer: NamingConventio
             .getSQL(ParamType.INLINED)
     }
 
-    protected fun castedField(
+    protected open fun castedField(
         field: Field<*>?,
         type: AirbyteType,
         alias: String?,
@@ -625,7 +625,7 @@ abstract class JdbcSqlGenerator(protected val namingTransformer: NamingConventio
         return DSL.cast(field, toDialectType(type))
     }
 
-    protected fun currentTimestamp(): Field<Timestamp> {
+    protected open fun currentTimestamp(): Field<Timestamp> {
         return DSL.currentTimestamp()
     }
 

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt

@@ -90,7 +90,7 @@ abstract class DestinationAcceptanceTest {
     protected var localRoot: Path? = null
     open protected var _testDataComparator: TestDataComparator = getTestDataComparator()
 
-    open fun getTestDataComparator(): TestDataComparator {
+    protected open fun getTestDataComparator(): TestDataComparator {
         return BasicTestDataComparator { this.resolveIdentifier(it) }
     }
 
@@ -102,7 +102,7 @@ abstract class DestinationAcceptanceTest {
          */
         get
 
-    protected fun supportsInDestinationNormalization(): Boolean {
+    protected open fun supportsInDestinationNormalization(): Boolean {
         return false
     }
 
@@ -208,7 +208,7 @@ abstract class DestinationAcceptanceTest {
     /**
      * Override to return true if a destination implements namespaces and should be tested as such.
      */
-    protected fun implementsNamespaces(): Boolean {
+    protected open fun implementsNamespaces(): Boolean {
         return false
     }
 
@@ -308,7 +308,7 @@ abstract class DestinationAcceptanceTest {
      * - can throw any exception, test framework will handle.
      */
     @Throws(Exception::class)
-    protected fun retrieveNormalizedRecords(
+    protected open fun retrieveNormalizedRecords(
         testEnv: TestDestinationEnv?,
         streamName: String?,
         namespace: String?
@@ -1036,7 +1036,7 @@ abstract class DestinationAcceptanceTest {
         }
     }
 
-    protected val maxRecordValueLimit: Int
+    protected open val maxRecordValueLimit: Int
         /** @return the max limit length allowed for values in the destination. */
         get() = 1000000000
 
@@ -1953,7 +1953,7 @@ abstract class DestinationAcceptanceTest {
         return false
     }
 
-    protected fun supportIncrementalSchemaChanges(): Boolean {
+    protected open fun supportIncrementalSchemaChanges(): Boolean {
         return false
     }
 

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/TestingNamespaces.kt

@@ -38,6 +38,7 @@ object TestingNamespaces {
      *
      * @return convention-compliant namespace
      */
+    @JvmStatic
     @JvmOverloads
     fun generate(prefix: String? = null): String {
         val userDefinedPrefix = if (prefix != null) prefix + "_" else ""
@@ -58,6 +59,7 @@ object TestingNamespaces {
      * @param namespace to check
      * @return true if the namespace is older than 2 days, otherwise false
      */
+    @JvmStatic
     fun isOlderThan2Days(namespace: String): Boolean {
         return isOlderThan(namespace, 2, ChronoUnit.DAYS)
     }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/comparator/AdvancedTestDataComparator.kt

@@ -24,7 +24,7 @@ open class AdvancedTestDataComparator : TestDataComparator {
         }
     }
 
-    protected fun resolveIdentifier(identifier: String?): List<String?> {
+    protected open fun resolveIdentifier(identifier: String?): List<String?> {
         return java.util.List.of(identifier)
     }
 
@@ -174,7 +174,7 @@ open class AdvancedTestDataComparator : TestDataComparator {
             .withZoneSameInstant(ZoneOffset.UTC)
     }
 
-    protected fun compareDateTimeWithTzValues(
+    protected open fun compareDateTimeWithTzValues(
         airbyteMessageValue: String,
         destinationValue: String
     ): Boolean {

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/typing_deduping/JdbcTypingDedupingTest.kt

@@ -36,15 +36,15 @@ abstract class JdbcTypingDedupingTest : BaseTypingDedupingTest() {
 
     protected abstract fun getDataSource(config: JsonNode?): DataSource?
 
-    protected val sourceOperations: JdbcCompatibleSourceOperations<*>
+    protected open val sourceOperations: JdbcCompatibleSourceOperations<*>
         /**
          * Subclasses may need to return a custom source operations if the default one does not
          * handle vendor-specific types correctly. For example, you most likely need to override
          * this method to deserialize JSON columns to JsonNode.
          */
         get() = JdbcUtils.defaultSourceOperations
 
-    protected val rawSchema: String
+    protected open val rawSchema: String
         /**
          * Subclasses using a config with a nonstandard raw table schema should override this
          * method.