[source-mysqlv2] A new mysqlv2 (#44606)

Co-authored-by: Marius Posta <marius@airbyte.io>

airbyte-integrations/connectors/source-mysql-v2/build.gradle

@@ -0,0 +1,23 @@
+plugins {
+    id 'airbyte-bulk-connector'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.source.mysql.MysqlSource'
+}
+
+airbyteBulkConnector {
+    core = 'extract'
+    toolkits = ['extract-jdbc']
+    cdk = 'local'
+}
+
+dependencies {
+    implementation 'mysql:mysql-connector-java:8.0.30'
+    implementation 'org.bouncycastle:bcpkix-jdk18on:1.77'
+    implementation 'org.bouncycastle:bcprov-jdk18on:1.77'
+    implementation 'org.bouncycastle:bctls-jdk18on:1.77'
+
+    testImplementation platform('org.testcontainers:testcontainers-bom:1.19.8')
+    testImplementation 'org.testcontainers:mysql'
+}

airbyte-integrations/connectors/source-mysql-v2/gradle.properties

@@ -0,0 +1,2 @@
+testExecutionConcurrency=1
+JunitMethodExecutionTimeout=5m

airbyte-integrations/connectors/source-mysql-v2/metadata.yaml

@@ -0,0 +1,28 @@
+data:
+  ab_internal:
+    ql: 200
+    sl: 100
+  allowedHosts:
+    hosts:
+      - ${host}
+      - ${tunnel_method.tunnel_host}
+  connectorSubtype: database
+  connectorType: source
+  definitionId: 561393ed-7e3a-4d0d-8b8b-90ded371754c
+  dockerImageTag: 0.0.1
+  dockerRepository: airbyte/source-mysql-v2
+  documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
+  githubIssueLabel: source-mysql-v2
+  icon: mysql.svg
+  license: ELv2
+  name: Mysqlv2 Source
+  registryOverrides:
+    cloud:
+      enabled: false
+    oss:
+      enabled: false
+  releaseStage: alpha
+  supportLevel: archived
+  tags:
+    - language:java
+metadataSpecVersion: "1.0"

airbyte-integrations/connectors/source-mysql-v2/src/main/kotlin/io/airbyte/integrations/source/mysql/MysqlJdbcSelectQuerier.kt

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.source.mysql
+
+import io.airbyte.cdk.read.JdbcSelectQuerier
+import io.airbyte.cdk.read.SelectQuerier
+import io.airbyte.cdk.read.SelectQuery
+import io.github.oshai.kotlinlogging.KotlinLogging
+import io.micronaut.context.annotation.Primary
+import javax.inject.Singleton
+
+private val log = KotlinLogging.logger {}
+
+@Singleton
+@Primary
+class MysqlJdbcSelectQuerier(val base: JdbcSelectQuerier) : SelectQuerier by base {
+
+    override fun executeQuery(
+        q: SelectQuery,
+        parameters: SelectQuerier.Parameters
+    ): SelectQuerier.Result {
+        log.info { "Executing query: ${q.sql}" }
+        return base.executeQuery(q, SelectQuerier.Parameters(fetchSize = Int.MIN_VALUE))
+    }
+}

airbyte-integrations/connectors/source-mysql-v2/src/main/kotlin/io/airbyte/integrations/source/mysql/MysqlSource.kt

@@ -0,0 +1,11 @@
+/* Copyright (c) 2024 Airbyte, Inc., all rights reserved. */
+package io.airbyte.integrations.source.mysql
+
+import io.airbyte.cdk.AirbyteSourceRunner
+
+object MysqlSource {
+    @JvmStatic
+    fun main(args: Array<String>) {
+        AirbyteSourceRunner.run(*args)
+    }
+}

airbyte-integrations/connectors/source-mysql-v2/src/main/kotlin/io/airbyte/integrations/source/mysql/MysqlSourceConfiguration.kt

@@ -0,0 +1,99 @@
+/* Copyright (c) 2024 Airbyte, Inc., all rights reserved. */
+package io.airbyte.integrations.source.mysql
+
+import io.airbyte.cdk.ConfigErrorException
+import io.airbyte.cdk.command.JdbcSourceConfiguration
+import io.airbyte.cdk.command.SourceConfiguration
+import io.airbyte.cdk.command.SourceConfigurationFactory
+import io.airbyte.cdk.ssh.SshConnectionOptions
+import io.airbyte.cdk.ssh.SshTunnelMethodConfiguration
+import io.github.oshai.kotlinlogging.KotlinLogging
+import jakarta.inject.Singleton
+import java.net.URLDecoder
+import java.nio.charset.StandardCharsets
+import java.time.Duration
+
+private val log = KotlinLogging.logger {}
+
+/** Mysql-specific implementation of [SourceConfiguration] */
+data class MysqlSourceConfiguration(
+    override val realHost: String,
+    override val realPort: Int,
+    override val sshTunnel: SshTunnelMethodConfiguration,
+    override val sshConnectionOptions: SshConnectionOptions,
+    override val jdbcUrlFmt: String,
+    override val jdbcProperties: Map<String, String>,
+    override val namespaces: Set<String>,
+    val cursorConfiguration: CursorConfiguration,
+    override val maxConcurrency: Int,
+    override val resourceAcquisitionHeartbeat: Duration = Duration.ofMillis(100L),
+    override val checkpointTargetInterval: Duration,
+    override val checkPrivileges: Boolean,
+) : JdbcSourceConfiguration {
+    override val global = cursorConfiguration is CdcCursor
+}
+
+@Singleton
+class MysqlSourceConfigurationFactory :
+    SourceConfigurationFactory<MysqlSourceConfigurationJsonObject, MysqlSourceConfiguration> {
+    override fun makeWithoutExceptionHandling(
+        pojo: MysqlSourceConfigurationJsonObject,
+    ): MysqlSourceConfiguration {
+        val realHost: String = pojo.host
+        val realPort: Int = pojo.port
+        val sshTunnel: SshTunnelMethodConfiguration = pojo.getTunnelMethodValue()
+        val jdbcProperties = mutableMapOf<String, String>()
+        jdbcProperties["user"] = pojo.username
+        pojo.password?.let { jdbcProperties["password"] = it }
+
+        // Parse URL parameters.
+        val pattern = "^([^=]+)=(.*)$".toRegex()
+        for (pair in (pojo.jdbcUrlParams ?: "").trim().split("&".toRegex())) {
+            if (pair.isBlank()) {
+                continue
+            }
+            val result: MatchResult? = pattern.matchEntire(pair)
+            if (result == null) {
+                log.warn { "ignoring invalid JDBC URL param '$pair'" }
+            } else {
+                val key: String = result.groupValues[1].trim()
+                val urlEncodedValue: String = result.groupValues[2].trim()
+                jdbcProperties[key] = URLDecoder.decode(urlEncodedValue, StandardCharsets.UTF_8)
+            }
+        }
+        // Determine protocol and configure encryption.
+        val encryption: Encryption = pojo.getEncryptionValue()
+        if (encryption is SslVerifyCertificate) {
+            // TODO: reuse JdbcSSLCOnnectionUtils; parse the input into properties
+        }
+        // Build JDBC URL
+        val address = "%s:%d"
+        val jdbcUrlFmt = "jdbc:mysql://${address}"
+        jdbcProperties["useCursorFetch"] = "true"
+        jdbcProperties["sessionVariables"] = "autocommit=0"
+        val defaultSchema: String = pojo.username.uppercase()
+        val sshOpts = SshConnectionOptions.fromAdditionalProperties(pojo.getAdditionalProperties())
+        val checkpointTargetInterval: Duration =
+            Duration.ofSeconds(pojo.checkpointTargetIntervalSeconds?.toLong() ?: 0)
+        if (!checkpointTargetInterval.isPositive) {
+            throw ConfigErrorException("Checkpoint Target Interval should be positive")
+        }
+        val maxConcurrency: Int = pojo.concurrency ?: 0
+        if ((pojo.concurrency ?: 0) <= 0) {
+            throw ConfigErrorException("Concurrency setting should be positive")
+        }
+        return MysqlSourceConfiguration(
+            realHost = realHost,
+            realPort = realPort,
+            sshTunnel = sshTunnel,
+            sshConnectionOptions = sshOpts,
+            jdbcUrlFmt = jdbcUrlFmt,
+            jdbcProperties = jdbcProperties,
+            namespaces = pojo.schemas?.toSet() ?: setOf(defaultSchema),
+            cursorConfiguration = pojo.getCursorConfigurationValue(),
+            checkpointTargetInterval = checkpointTargetInterval,
+            maxConcurrency = maxConcurrency,
+            checkPrivileges = pojo.checkPrivileges ?: true,
+        )
+    }
+}