feat: Add Destination MSSQL v2 rc.

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/write/BasicFunctionalityIntegrationTest.kt

@@ -193,8 +193,10 @@ abstract class BasicFunctionalityIntegrationTest(
         configUpdater = configUpdater,
         envVars = envVars,
     ) {
-    val parsedConfig =
-        ValidatedJsonUtils.parseOne(configSpecClass, configUpdater.update(configContents))
+
+    // Update config with any replacements.  This may be necessary when using testcontainers.
+    val configAsString = configUpdater.update(configContents)
+    val parsedConfig = ValidatedJsonUtils.parseOne(configSpecClass, configAsString)
 
     @Test
     open fun testBasicWrite() {
@@ -209,7 +211,7 @@ abstract class BasicFunctionalityIntegrationTest(
             )
         val messages =
             runSync(
-                configContents,
+                configAsString,
                 stream,
                 listOf(
                     InputRecord(
@@ -260,7 +262,7 @@ abstract class BasicFunctionalityIntegrationTest(
             {
                 if (verifyDataWriting) {
                     dumpAndDiffRecords(
-                        ValidatedJsonUtils.parseOne(configSpecClass, configContents),
+                        ValidatedJsonUtils.parseOne(configSpecClass, configAsString),
                         listOf(
                             OutputRecord(
                                 extractedAt = 1234,
@@ -321,7 +323,7 @@ abstract class BasicFunctionalityIntegrationTest(
 
         val messages =
             runSync(
-                configContents,
+                configAsString,
                 stream,
                 listOf(
                     InputFile(
@@ -359,7 +361,7 @@ abstract class BasicFunctionalityIntegrationTest(
             )
         })
 
-        val config = ValidatedJsonUtils.parseOne(configSpecClass, configContents)
+        val config = ValidatedJsonUtils.parseOne(configSpecClass, configAsString)
         val fileContent = dataDumper.dumpFile(config, stream)
 
         assertEquals(listOf("123"), fileContent)
@@ -380,7 +382,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 )
             val stateMessage =
                 runSyncUntilStateAck(
-                    configContents,
+                    this@BasicFunctionalityIntegrationTest.configContents,
                     stream,
                     listOf(
                         InputRecord(
@@ -398,7 +400,7 @@ abstract class BasicFunctionalityIntegrationTest(
                     ),
                     allowGracefulShutdown = false,
                 )
-            runSync(configContents, stream, emptyList())
+            runSync(this@BasicFunctionalityIntegrationTest.configContents, stream, emptyList())
 
             val streamName = stateMessage.stream.streamDescriptor.name
             val streamNamespace = stateMessage.stream.streamDescriptor.namespace
@@ -461,7 +463,7 @@ abstract class BasicFunctionalityIntegrationTest(
         val stream1 = makeStream(randomizedNamespace + "_1")
         val stream2 = makeStream(randomizedNamespace + "_2")
         runSync(
-            configContents,
+            configAsString,
             DestinationCatalog(
                 listOf(
                     stream1,
@@ -590,7 +592,7 @@ abstract class BasicFunctionalityIntegrationTest(
                     serialized = "",
                 )
             }
-        runSync(configContents, catalog, messages)
+        runSync(configAsString, catalog, messages)
         assertAll(
             catalog.streams.map { stream ->
                 {
@@ -630,7 +632,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId,
             )
         runSync(
-            configContents,
+            configAsString,
             makeStream(generationId = 12, minimumGenerationId = 0, syncId = 42),
             listOf(
                 InputRecord(
@@ -643,7 +645,7 @@ abstract class BasicFunctionalityIntegrationTest(
         )
         val finalStream = makeStream(generationId = 13, minimumGenerationId = 13, syncId = 43)
         runSync(
-            configContents,
+            configAsString,
             finalStream,
             listOf(
                 InputRecord(
@@ -725,7 +727,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 41,
             )
         runSync(
-            configContents,
+            configAsString,
             stream1,
             listOf(
                 makeInputRecord(1, "2024-01-23T01:00:00Z", 100),
@@ -764,7 +766,7 @@ abstract class BasicFunctionalityIntegrationTest(
             )
         // Run a sync, but emit a status incomplete. This should not delete any existing data.
         runSyncUntilStateAck(
-            configContents,
+            configAsString,
             stream2,
             listOf(makeInputRecord(1, "2024-01-23T02:00:00Z", 200)),
             StreamCheckpoint(
@@ -813,7 +815,7 @@ abstract class BasicFunctionalityIntegrationTest(
         // Run a third sync, this time with a successful status.
         // This should delete the first sync's data, and retain the second+third syncs' data.
         runSync(
-            configContents,
+            configAsString,
             stream2,
             listOf(makeInputRecord(2, "2024-01-23T03:00:00Z", 300)),
         )
@@ -894,7 +896,7 @@ abstract class BasicFunctionalityIntegrationTest(
             )
         // Run a sync, but emit a stream status incomplete.
         runSyncUntilStateAck(
-            configContents,
+            configAsString,
             stream,
             listOf(makeInputRecord(1, "2024-01-23T02:00:00Z", 200)),
             StreamCheckpoint(
@@ -929,7 +931,7 @@ abstract class BasicFunctionalityIntegrationTest(
         // Run a second sync, this time with a successful status.
         // This should retain the first syncs' data.
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(makeInputRecord(2, "2024-01-23T03:00:00Z", 300)),
         )
@@ -1015,7 +1017,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 41,
             )
         runSync(
-            configContents,
+            configAsString,
             stream1,
             listOf(
                 makeInputRecord(1, "2024-01-23T01:00:00Z", 100),
@@ -1055,7 +1057,7 @@ abstract class BasicFunctionalityIntegrationTest(
         // Run a sync, but emit a stream status incomplete. This should not delete any existing
         // data.
         runSyncUntilStateAck(
-            configContents,
+            configAsString,
             stream2,
             listOf(makeInputRecord(1, "2024-01-23T02:00:00Z", 200)),
             StreamCheckpoint(
@@ -1110,7 +1112,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 43,
             )
         runSync(
-            configContents,
+            configAsString,
             stream3,
             listOf(makeInputRecord(2, "2024-01-23T03:00:00Z", 300)),
         )
@@ -1170,7 +1172,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId,
             )
         runSync(
-            configContents,
+            configAsString,
             makeStream(syncId = 42),
             listOf(
                 InputRecord(
@@ -1183,7 +1185,7 @@ abstract class BasicFunctionalityIntegrationTest(
         )
         val finalStream = makeStream(syncId = 43)
         runSync(
-            configContents,
+            configAsString,
             finalStream,
             listOf(
                 InputRecord(
@@ -1236,7 +1238,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId,
             )
         runSync(
-            configContents,
+            configAsString,
             makeStream(
                 syncId = 42,
                 linkedMapOf("id" to intType, "to_drop" to stringType, "to_change" to intType)
@@ -1256,7 +1258,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 linkedMapOf("id" to intType, "to_change" to stringType, "to_add" to stringType)
             )
         runSync(
-            configContents,
+            configAsString,
             finalStream,
             listOf(
                 InputRecord(
@@ -1330,7 +1332,7 @@ abstract class BasicFunctionalityIntegrationTest(
 
         val sync1Stream = makeStream(syncId = 42)
         runSync(
-            configContents,
+            configAsString,
             sync1Stream,
             listOf(
                 // emitted_at:1000 is equal to 1970-01-01 00:00:01Z.
@@ -1391,7 +1393,7 @@ abstract class BasicFunctionalityIntegrationTest(
 
         val sync2Stream = makeStream(syncId = 43)
         runSync(
-            configContents,
+            configAsString,
             sync2Stream,
             listOf(
                 // Update both Alice and Bob
@@ -1475,9 +1477,9 @@ abstract class BasicFunctionalityIntegrationTest(
                 // instead of being able to fallback onto extractedAt.
                 emittedAtMs = 100,
             )
-        runSync(configContents, makeStream("cursor1"), listOf(makeRecord("cursor1")))
+        runSync(configAsString, makeStream("cursor1"), listOf(makeRecord("cursor1")))
         val stream2 = makeStream("cursor2")
-        runSync(configContents, stream2, listOf(makeRecord("cursor2")))
+        runSync(configAsString, stream2, listOf(makeRecord("cursor2")))
         dumpAndDiffRecords(
             parsedConfig,
             listOf(
@@ -1534,7 +1536,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 )
             }
         // Just verify that we don't crash.
-        assertDoesNotThrow { runSync(configContents, DestinationCatalog(streams), messages) }
+        assertDoesNotThrow { runSync(configAsString, DestinationCatalog(streams), messages) }
     }
 
     /**
@@ -1587,7 +1589,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 emittedAtMs = 100,
             )
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(
                 // A record with valid values for all fields
@@ -1871,7 +1873,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 42,
             )
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(
                 InputRecord(
@@ -2050,7 +2052,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 42,
             )
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(
                 InputRecord(
@@ -2221,7 +2223,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 42,
             )
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(
                 InputRecord(
@@ -2413,7 +2415,7 @@ abstract class BasicFunctionalityIntegrationTest(
                 syncId = 42,
             )
         runSync(
-            configContents,
+            configAsString,
             stream,
             listOf(
                 InputRecord(

airbyte-integrations/connectors/destination-mssql-v2/build.gradle.kts

@@ -22,7 +22,7 @@ application {
     //applicationDefaultJvmArgs = listOf("-XX:+ExitOnOutOfMemoryError", "-XX:MaxRAMPercentage=75.0", "--add-opens", "java.base/sun.nio.ch=ALL-UNNAMED", "--add-opens", "java.base/sun.security.action=ALL-UNNAMED", "--add-opens", "java.base/java.lang=ALL-UNNAMED")
 }
 
-val junitVersion = "5.11.3"
+val junitVersion = "5.11.4"
 
 configurations.configureEach {
     // Exclude additional SLF4J providers from all classpaths
@@ -38,13 +38,16 @@ dependencies {
     implementation("com.microsoft.sqlserver:mssql-jdbc:12.8.1.jre11")
     implementation("io.github.oshai:kotlin-logging-jvm:7.0.0")
     implementation("jakarta.inject:jakarta.inject-api:2.0.1")
-    implementation("com.github.spotbugs:spotbugs-annotations:4.8.6")
-    implementation("io.micronaut:micronaut-inject:4.6.1")
+    implementation("com.github.spotbugs:spotbugs-annotations:4.9.0")
+    implementation("io.micronaut:micronaut-inject:4.7.12")
+    implementation("com.zaxxer:HikariCP:6.2.1")
 
-    testImplementation("io.mockk:mockk:1.13.13")
+    testImplementation("io.mockk:mockk:1.13.16")
     testImplementation("org.junit.jupiter:junit-jupiter-api:$junitVersion")
     testImplementation("org.junit.jupiter:junit-jupiter-params:$junitVersion")
     testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:$junitVersion")
+
+    integrationTestImplementation("org.testcontainers:mssqlserver:1.20.4")
 }
 
 tasks.named<Test>("test") {

airbyte-integrations/connectors/destination-mssql-v2/docker-compose.yml

@@ -0,0 +1,8 @@
+services:
+  sql-server:
+    image: mcr.microsoft.com/mssql/server:2022-latest
+    ports:
+      - "1433:1433"
+    environment:
+      - ACCEPT_EULA=Y
+      - MSSQL_SA_PASSWORD=A_Str0ng_Required_Password

airbyte-integrations/connectors/destination-mssql-v2/metadata.yaml

@@ -16,7 +16,7 @@ data:
             type: GSM
   connectorType: destination
   definitionId: 37a928c1-2d5c-431a-a97d-ae236bd1ea0c
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.3
   dockerRepository: airbyte/destination-mssql-v2
   documentationUrl: https://docs.airbyte.com/integrations/destinations/mssql-v2
   githubIssueLabel: destination-mssql-v2

airbyte-integrations/connectors/destination-mssql-v2/src/main/kotlin/io/airbyte/integrations/destination/mssql/v2/MSSQLChecker.kt

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.mssql.v2
+
+import io.airbyte.cdk.load.check.DestinationChecker
+import io.airbyte.integrations.destination.mssql.v2.config.MSSQLConfiguration
+import io.airbyte.integrations.destination.mssql.v2.config.MSSQLDataSourceFactory
+import jakarta.inject.Singleton
+import java.util.UUID
+import javax.sql.DataSource
+
+@Singleton
+class MSSQLChecker(private val dataSourceFactory: MSSQLDataSourceFactory) :
+    DestinationChecker<MSSQLConfiguration> {
+    override fun check(config: MSSQLConfiguration) {
+        val dataSource: DataSource = dataSourceFactory.getDataSource(config)
+        val testTableName = "check_test_${UUID.randomUUID()}"
+        val fullyQualifiedTableName = "[${config.rawDataSchema}].[${testTableName}]"
+        dataSource.connection.use { connection ->
+            connection.createStatement().use { statement ->
+                statement.executeUpdate(
+                    """
+                    CREATE TABLE ${fullyQualifiedTableName} (test int);
+                    DROP TABLE ${fullyQualifiedTableName};
+                    """.trimIndent(),
+                )
+            }
+        }
+    }
+}