[Source-postgres] : Move to new Kotlin CDK

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

@@ -13,6 +13,6 @@ import com.fasterxml.jackson.databind.JsonNode
  * accidentally closing a shared resource.
  */
 open class AbstractDatabase {
-    @JvmField var sourceConfig: JsonNode? = null
-    @JvmField var databaseConfig: JsonNode? = null
+    var sourceConfig: JsonNode? = null
+    var databaseConfig: JsonNode? = null
 }

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

@@ -13,6 +13,7 @@ import io.airbyte.protocol.models.v0.AirbyteAnalyticsTraceMessage
 object DbAnalyticsUtils {
     const val CDC_CURSOR_INVALID_KEY: String = "db-sources-cdc-cursor-invalid"
 
+    @JvmStatic
     fun cdcCursorInvalidMessage(): AirbyteAnalyticsTraceMessage {
         return AirbyteAnalyticsTraceMessage().withType(CDC_CURSOR_INVALID_KEY).withValue("1")
     }

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

@@ -34,6 +34,7 @@ enum class DatabaseDriver(val driverClassName: String, val urlFormatString: Stri
          * @param driverClassName The driver class name.
          * @return The matching [DatabaseDriver] enumerated value or `null` if no match is found.
          */
+        @JvmStatic
         fun findByDriverClassName(driverClassName: String?): DatabaseDriver {
             lateinit var selected: DatabaseDriver
 

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

@@ -98,7 +98,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putBoolean(
+    protected open fun putBoolean(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,
@@ -156,7 +156,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putDouble(
+    protected open fun putDouble(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,
@@ -188,7 +188,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putBigDecimal(
+    protected open fun putBigDecimal(
         node: ObjectNode,
         columnName: String?,
         resultSet: ResultSet,
@@ -208,7 +208,12 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun putDate(node: ObjectNode, columnName: String?, resultSet: ResultSet, index: Int) {
+    protected open fun putDate(
+        node: ObjectNode,
+        columnName: String?,
+        resultSet: ResultSet,
+        index: Int
+    ) {
         node.put(columnName, resultSet.getString(index))
     }
 
@@ -280,7 +285,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun setTimestamp(
+    protected open fun setTimestamp(
         preparedStatement: PreparedStatement,
         parameterIndex: Int,
         value: String?
@@ -293,7 +298,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
     }
 
     @Throws(SQLException::class)
-    protected fun setDate(
+    protected open fun setDate(
         preparedStatement: PreparedStatement,
         parameterIndex: Int,
         value: String

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/JdbcConnector.kt

@@ -36,6 +36,7 @@ protected constructor(@JvmField protected val driverClassName: String) : BaseCon
          * @param driverClassName name of the JDBC driver
          * @return DataSourceBuilder class used to create dynamic fields for DataSource
          */
+        @JvmStatic
         fun getConnectionTimeout(
             connectionProperties: Map<String, String>,
             driverClassName: String?

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

@@ -33,6 +33,7 @@ object AirbyteTraceMessageUtility {
         )
     }
 
+    @JvmStatic
     fun emitEstimateTrace(
         byteEstimate: Long,
         type: AirbyteEstimateTraceMessage.Type?,
@@ -55,10 +56,12 @@ object AirbyteTraceMessageUtility {
         )
     }
 
+    @JvmStatic
     fun emitAnalyticsTrace(airbyteAnalyticsTraceMessage: AirbyteAnalyticsTraceMessage) {
         emitMessage(makeAnalyticsTraceAirbyteMessage(airbyteAnalyticsTraceMessage))
     }
 
+    @JvmStatic
     fun emitErrorTrace(
         e: Throwable,
         displayMessage: String?,

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

@@ -17,6 +17,7 @@ object SshHelpers {
         get() = getSpecAndInjectSsh(Optional.empty())
 
     @Throws(IOException::class)
+    @JvmStatic
     fun getSpecAndInjectSsh(group: Optional<String>): ConnectorSpecification? {
         val originalSpec =
             Jsons.deserialize(
@@ -28,6 +29,7 @@ object SshHelpers {
 
     @JvmOverloads
     @Throws(IOException::class)
+    @JvmStatic
     fun injectSshIntoSpec(
         connectorSpecification: ConnectorSpecification,
         group: Optional<String> = Optional.empty()

airbyte-cdk/java/airbyte-cdk/core/src/testFixtures/kotlin/io/airbyte/cdk/integrations/base/ssh/SshBastionContainer.kt

@@ -134,6 +134,7 @@ class SshBastionContainer : AutoCloseable {
         private val SSH_USER: String? = "sshuser"
         private val SSH_PASSWORD: String? = "secret"
 
+        @JvmStatic
         /**
          * Returns the inner docker network ip address and port of a container. This can be used to
          * reach a container from another container running on the same network
@@ -157,6 +158,7 @@ class SshBastionContainer : AutoCloseable {
             )
         }
 
+        @JvmStatic
         /**
          * Returns the outer docker network ip address and port of a container. This can be used to
          * reach a container from the host machine

airbyte-cdk/java/airbyte-cdk/core/src/testFixtures/kotlin/io/airbyte/cdk/testutils/TestDatabase.kt

@@ -302,10 +302,7 @@ protected constructor(val container: C) : AutoCloseable {
         }
 
         companion object {
-            private val DEFAULT_CDC_REPLICATION_INITIAL_WAIT: Duration = Duration.ofSeconds(5)
-            fun getDefaultCdcReplicationInitialWait(): Duration {
-                return DEFAULT_CDC_REPLICATION_INITIAL_WAIT
-            }
+            @JvmField val DEFAULT_CDC_REPLICATION_INITIAL_WAIT: Duration? = Duration.ofSeconds(5)
         }
     }
 

airbyte-cdk/java/airbyte-cdk/datastore-postgres/src/main/kotlin/io/airbyte/cdk/db/PgLsn.kt

@@ -35,6 +35,7 @@ class PgLsn private constructor(private val lsn: Long) : Comparable<PgLsn> {
             return PgLsn(lsn)
         }
 
+        @JvmStatic
         fun fromPgString(lsn: String): PgLsn {
             return PgLsn(lsnToLong(lsn))
         }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/AirbyteDebeziumHandler.kt

@@ -154,6 +154,7 @@ class AirbyteDebeziumHandler<T>(
          */
         const val QUEUE_CAPACITY: Int = 10000
 
+        @JvmStatic
         fun isAnyStreamIncrementalSyncMode(catalog: ConfiguredAirbyteCatalog): Boolean {
             return catalog.streams
                 .stream()

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/CdcTargetPosition.kt

@@ -70,7 +70,7 @@ interface CdcTargetPosition<T> {
      * @return Returns `true` if both offsets are at the same position. Otherwise, it returns
      * `false`
      */
-    fun isSameOffset(offsetA: Map<*, *>, offsetB: Map<String, String>): Boolean {
+    fun isSameOffset(offsetA: Map<String, String>, offsetB: Map<String, String>): Boolean {
         return false
     }
 }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/DebeziumIteratorConstants.kt

@@ -12,6 +12,6 @@ object DebeziumIteratorConstants {
     // TODO: Move these variables to a separate class IteratorConstants, as they will be used in
     // state
     // iterators for non debezium cases too.
-    val SYNC_CHECKPOINT_DURATION: Duration = Duration.ofMinutes(15)
+    @JvmField val SYNC_CHECKPOINT_DURATION: Duration = Duration.ofMinutes(15)
     const val SYNC_CHECKPOINT_RECORDS: Int = 10000
 }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/AirbyteFileOffsetBackingStore.kt

@@ -193,6 +193,7 @@ class AirbyteFileOffsetBackingStore(
             return ByteBuffer.wrap(s!!.toByteArray(StandardCharsets.UTF_8))
         }
 
+        @JvmStatic
         fun initializeState(
             cdcState: JsonNode?,
             dbName: Optional<String>

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/DebeziumConverterUtils.kt

@@ -24,6 +24,7 @@ class DebeziumConverterUtils private constructor() {
         private val LOGGER: Logger = LoggerFactory.getLogger(DebeziumConverterUtils::class.java)
 
         /** TODO : Replace usage of this method with [io.airbyte.cdk.db.jdbc.DateTimeConverter] */
+        @JvmStatic
         fun convertDate(input: Any): String {
             /**
              * While building this custom converter we were not sure what type debezium could return
@@ -61,6 +62,7 @@ class DebeziumConverterUtils private constructor() {
             return input.toString()
         }
 
+        @JvmStatic
         fun convertDefaultValue(field: RelationalColumn): Any? {
             if (field.isOptional) {
                 return null

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.kt

@@ -38,7 +38,7 @@ class DebeziumMessageProducer<T>(
      * generating an unneeded usage of networking and processing.
      */
     private val initialOffset: HashMap<String, String>
-    private val previousCheckpointOffset: HashMap<String?, String?>
+    private val previousCheckpointOffset: HashMap<String, String>
     private val offsetManager: AirbyteFileOffsetBackingStore?
     private val targetPosition: CdcTargetPosition<T>
     private val schemaHistoryManager: Optional<AirbyteSchemaHistoryStorage>
@@ -55,7 +55,7 @@ class DebeziumMessageProducer<T>(
             throw RuntimeException("Offset manager cannot be null")
         }
         this.schemaHistoryManager = schemaHistoryManager
-        this.previousCheckpointOffset = offsetManager.read() as HashMap<String?, String?>
+        this.previousCheckpointOffset = offsetManager.read() as HashMap<String, String>
         this.initialOffset = HashMap(this.previousCheckpointOffset)
     }
 

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/AbstractJdbcSource.kt

@@ -83,13 +83,13 @@ import org.slf4j.LoggerFactory
 @SuppressFBWarnings("NP_PARAMETER_MUST_BE_NONNULL_BUT_MARKED_AS_NULLABLE")
 abstract class AbstractJdbcSource<Datatype>(
     driverClass: String,
-    protected val streamingQueryConfigProvider: Supplier<JdbcStreamingQueryConfig>,
+    @JvmField val streamingQueryConfigProvider: Supplier<JdbcStreamingQueryConfig>,
     sourceOperations: JdbcCompatibleSourceOperations<Datatype>
 ) : AbstractDbSource<Datatype, JdbcDatabase>(driverClass), Source {
-    protected val sourceOperations: JdbcCompatibleSourceOperations<Datatype>
+    @JvmField val sourceOperations: JdbcCompatibleSourceOperations<Datatype>
 
     override var quoteString: String? = null
-    protected var dataSources: MutableCollection<DataSource> = ArrayList()
+    @JvmField val dataSources: MutableCollection<DataSource> = ArrayList()
 
     init {
         this.sourceOperations = sourceOperations
@@ -304,7 +304,10 @@ abstract class AbstractJdbcSource<Datatype>(
 
     // needs to override isNotInternalSchema for connectors that override
     // getPrivilegesTableForCurrentUser()
-    protected fun isNotInternalSchema(jsonNode: JsonNode, internalSchemas: Set<String?>): Boolean {
+    protected open fun isNotInternalSchema(
+        jsonNode: JsonNode,
+        internalSchemas: Set<String?>
+    ): Boolean {
         return !internalSchemas.contains(jsonNode.get(INTERNAL_SCHEMA_NAME).asText())
     }
 

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/JdbcDataSourceUtils.kt

@@ -18,6 +18,7 @@ object JdbcDataSourceUtils {
      * @param defaultParameters connection properties map as specified by each Jdbc source
      * @throws IllegalArgumentException
      */
+    @JvmStatic
     fun assertCustomParametersDontOverwriteDefaultParameters(
         customParameters: Map<String, String>,
         defaultParameters: Map<String, String>
@@ -57,6 +58,7 @@ object JdbcDataSourceUtils {
      * @param config A configuration used to check Jdbc connection
      * @return A mapping of the default connection properties
      */
+    @JvmStatic
     fun getDefaultConnectionProperties(config: JsonNode): Map<String, String> {
         // NOTE that Postgres returns an empty map for some reason?
         return parseJdbcParameters(

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/JdbcSSLConnectionUtils.kt

@@ -71,6 +71,7 @@ class JdbcSSLConnectionUtils {
          * @return map containing relevant parsed values including location of keystore or an empty
          * map
          */
+        @JvmStatic
         fun parseSSLConfig(config: JsonNode): Map<String, String> {
             LOGGER.debug("source config: {}", config)
 
@@ -143,6 +144,7 @@ class JdbcSSLConnectionUtils {
             return additionalParameters
         }
 
+        @JvmStatic
         fun prepareCACertificateKeyStore(config: JsonNode): Pair<URI, String>? {
             // if config available
             // if has CA cert - make keystore
@@ -266,6 +268,7 @@ class JdbcSSLConnectionUtils {
             return clientCertKeyStorePair
         }
 
+        @JvmStatic
         fun fileFromCertPem(certPem: String?): Path {
             try {
                 val path = Files.createTempFile(null, ".crt")

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/dto/JdbcPrivilegeDto.kt

@@ -79,6 +79,7 @@ class JdbcPrivilegeDto(
     }
 
     companion object {
+        @JvmStatic
         fun builder(): JdbcPrivilegeDtoBuilder {
             return JdbcPrivilegeDtoBuilder()
         }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/AbstractDbSource.kt

@@ -52,7 +52,7 @@ abstract class AbstractDbSource<DataType, Database : AbstractDatabase?>
 protected constructor(driverClassName: String) :
     JdbcConnector(driverClassName), Source, AutoCloseable {
     // TODO: Remove when the flag is not use anymore
-    protected var featureFlags: FeatureFlags = EnvVariableFeatureFlags()
+    var featureFlags: FeatureFlags = EnvVariableFeatureFlags()
 
     @Trace(operationName = CHECK_TRACE_OPERATION_NAME)
     @Throws(Exception::class)
@@ -270,7 +270,7 @@ protected constructor(driverClassName: String) :
      * @throws SQLException exception
      */
     @Throws(SQLException::class)
-    protected fun verifyCursorColumnValues(
+    protected open fun verifyCursorColumnValues(
         database: Database,
         schema: String?,
         tableName: String?,
@@ -286,7 +286,7 @@ protected constructor(driverClassName: String) :
      *
      * @param database database
      */
-    protected fun estimateFullRefreshSyncSize(
+    protected open fun estimateFullRefreshSyncSize(
         database: Database,
         configuredAirbyteStream: ConfiguredAirbyteStream?
     ) {
@@ -327,7 +327,7 @@ protected constructor(driverClassName: String) :
         )
     }
 
-    protected fun getIncrementalIterators(
+    protected open fun getIncrementalIterators(
         database: Database,
         catalog: ConfiguredAirbyteCatalog,
         tableNameToTable: Map<String?, TableInfo<CommonField<DataType>>>,
@@ -601,7 +601,7 @@ protected constructor(driverClassName: String) :
      * Oracle DB - the schema is the user, you cannot REVOKE a privilege on a table from its owner).
      */
     @Throws(SQLException::class)
-    protected fun <T> getPrivilegesTableForCurrentUser(
+    protected open fun <T> getPrivilegesTableForCurrentUser(
         database: JdbcDatabase?,
         schema: String?
     ): Set<T> {
@@ -670,7 +670,7 @@ protected constructor(driverClassName: String) :
          */
         get
 
-    protected val excludedViews: Set<String>
+    protected open val excludedViews: Set<String>
         /**
          * Get list of system views in order to exclude them from the `discover` result list.
          *
@@ -765,7 +765,7 @@ protected constructor(driverClassName: String) :
         cursorFieldType: DataType
     ): AutoCloseableIterator<AirbyteRecordData>
 
-    protected val stateEmissionFrequency: Int
+    protected open val stateEmissionFrequency: Int
         /**
          * When larger than 0, the incremental iterator will emit intermediate state for every N
          * records. Please note that if intermediate state emission is enabled, the incremental

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/DbSourceDiscoverUtil.kt

@@ -147,6 +147,7 @@ object DbSourceDiscoverUtil {
         return AirbyteCatalog().withStreams(streams)
     }
 
+    @JvmStatic
     fun getFullyQualifiedTableName(nameSpace: String?, tableName: String): String {
         return if (nameSpace != null) "$nameSpace.$tableName" else tableName
     }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/RelationalDbQueryUtils.kt

@@ -28,6 +28,7 @@ object RelationalDbQueryUtils {
         }
     }
 
+    @JvmStatic
     fun enquoteIdentifierList(identifiers: List<String>, quoteString: String): String {
         val joiner = StringJoiner(",")
         for (identifier in identifiers) {
@@ -37,6 +38,7 @@ object RelationalDbQueryUtils {
     }
 
     /** @return fully qualified table name with the schema (if a schema exists) in quotes. */
+    @JvmStatic
     fun getFullyQualifiedTableNameWithQuoting(
         nameSpace: String?,
         tableName: String,