fix compiler errors

stephane-airbyte · stephane-airbyte · commit 4b47359760aa · 2024-03-27T02:32:27.000Z
diff --git a/airbyte-cdk/java/airbyte-cdk/azure-destinations/build.gradle b/airbyte-cdk/java/airbyte-cdk/azure-destinations/build.gradle
@@ -1,3 +1,9 @@
+compileKotlin {
+    compilerOptions {
+        allWarningsAsErrors = false
+    }
+}
+
 dependencies {
     implementation project(':airbyte-cdk:java:airbyte-cdk:dependencies')
     implementation project(':airbyte-cdk:java:airbyte-cdk:core')
diff --git a/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageConfig.kt b/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageConfig.kt
@@ -6,27 +6,28 @@ package io.airbyte.cdk.integrations.destination.jdbc.copy.azure
 import com.fasterxml.jackson.databind.JsonNode
 import java.util.*
 
-class AzureBlobStorageConfig(private val endpointDomainName: String?,
-                             val accountName: String,
-                             val containerName: String,
-                             val sasToken: String) {
-    fun getEndpointDomainName(): String {
-        return endpointDomainName ?: DEFAULT_STORAGE_ENDPOINT_DOMAIN_NAME
-    }
+class AzureBlobStorageConfig(
+    val endpointDomainName: String = DEFAULT_STORAGE_ENDPOINT_DOMAIN_NAME,
+    val accountName: String,
+    val containerName: String,
+    val sasToken: String
+) {
 
     val endpointUrl: String
-        get() = String.format(Locale.ROOT, "https://%s.%s", accountName, getEndpointDomainName())
+        get() = String.format(Locale.ROOT, "https://%s.%s", accountName, endpointDomainName)
 
     companion object {
         private const val DEFAULT_STORAGE_ENDPOINT_DOMAIN_NAME = "blob.core.windows.net"
 
         fun getAzureBlobConfig(config: JsonNode): AzureBlobStorageConfig {
             return AzureBlobStorageConfig(
-                    if (config["azure_blob_storage_endpoint_domain_name"] == null) DEFAULT_STORAGE_ENDPOINT_DOMAIN_NAME
-                    else config["azure_blob_storage_endpoint_domain_name"].asText(),
-                    config["azure_blob_storage_account_name"].asText(),
-                    config["azure_blob_storage_container_name"].asText(),
-                    config["azure_blob_storage_sas_token"].asText())
+                if (config["azure_blob_storage_endpoint_domain_name"] == null)
+                    DEFAULT_STORAGE_ENDPOINT_DOMAIN_NAME
+                else config["azure_blob_storage_endpoint_domain_name"].asText(),
+                config["azure_blob_storage_account_name"].asText(),
+                config["azure_blob_storage_container_name"].asText(),
+                config["azure_blob_storage_sas_token"].asText()
+            )
         }
     }
 }
diff --git a/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageStreamCopier.kt b/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageStreamCopier.kt
@@ -17,65 +17,70 @@ import io.airbyte.cdk.integrations.destination.jdbc.copy.StreamCopier
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.AirbyteRecordMessage
 import io.airbyte.protocol.models.v0.DestinationSyncMode
-import org.apache.commons.csv.CSVFormat
-import org.apache.commons.csv.CSVPrinter
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
 import java.io.*
 import java.nio.charset.StandardCharsets
 import java.sql.SQLException
 import java.sql.Timestamp
 import java.time.Instant
 import java.util.*
 import java.util.function.Consumer
+import org.apache.commons.csv.CSVFormat
+import org.apache.commons.csv.CSVPrinter
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+abstract class AzureBlobStorageStreamCopier(
+    protected val stagingFolder: String,
+    private val destSyncMode: DestinationSyncMode,
+    protected val schemaName: String,
+    protected val streamName: String,
+    private val specializedBlobClientBuilder: SpecializedBlobClientBuilder,
+    protected val db: JdbcDatabase,
+    protected val azureBlobConfig: AzureBlobStorageConfig,
+    private val nameTransformer: StandardNameTransformer,
+    private val sqlOperations: SqlOperations
+) : StreamCopier {
+    protected var filenameGenerator: StagingFilenameGenerator =
+        StagingFilenameGenerator(
+            streamName,
+            GlobalDataSizeConstants.DEFAULT_MAX_BATCH_SIZE_BYTES.toLong()
+        )
+    protected val azureStagingFiles: MutableSet<String> = HashSet()
 
-abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
-                                            private val destSyncMode: DestinationSyncMode,
-                                            protected val schemaName: String,
-                                            protected val streamName: String,
-                                            private val specializedBlobClientBuilder: SpecializedBlobClientBuilder,
-                                            protected val db: JdbcDatabase,
-                                            protected val azureBlobConfig: AzureBlobStorageConfig,
-                                            private val nameTransformer: StandardNameTransformer,
-                                            private val sqlOperations: SqlOperations) : StreamCopier {
-    protected var filenameGenerator: StagingFilenameGenerator = StagingFilenameGenerator(streamName, GlobalDataSizeConstants.DEFAULT_MAX_BATCH_SIZE_BYTES.toLong())
-    protected val azureStagingFiles: MutableSet<String?> = HashSet()
-
-    @get:VisibleForTesting
-    val tmpTableName: String = nameTransformer.getTmpTableName(streamName)
+    @get:VisibleForTesting val tmpTableName: String = nameTransformer.getTmpTableName(streamName)
     protected val activeStagingWriterFileNames: MutableSet<String?> = HashSet()
     private val csvPrinters = HashMap<String?, CSVPrinter>()
     private val blobClients = HashMap<String?, AppendBlobClient>()
     override var currentFile: String? = null
-        private set
-
-    fun getAzureStagingFiles(): Set<String?> {
-        return azureStagingFiles
-    }
 
     @Throws(Exception::class)
     override fun write(id: UUID?, recordMessage: AirbyteRecordMessage?, azureFileName: String?) {
         if (csvPrinters.containsKey(azureFileName)) {
-            csvPrinters[azureFileName]!!.printRecord(id,
-                    Jsons.serialize(recordMessage!!.data),
-                    Timestamp.from(Instant.ofEpochMilli(recordMessage.emittedAt)))
+            csvPrinters[azureFileName]!!.printRecord(
+                id,
+                Jsons.serialize(recordMessage!!.data),
+                Timestamp.from(Instant.ofEpochMilli(recordMessage.emittedAt))
+            )
         }
     }
 
     override fun prepareStagingFile(): String? {
         currentFile = prepareAzureStagingFile()
+        val currentFile = this.currentFile!!
         if (!azureStagingFiles.contains(currentFile)) {
             azureStagingFiles.add(currentFile)
             activeStagingWriterFileNames.add(currentFile)
 
-            val appendBlobClient = specializedBlobClientBuilder
-                    .blobName(currentFile)
-                    .buildAppendBlobClient()
+            val appendBlobClient =
+                specializedBlobClientBuilder.blobName(currentFile).buildAppendBlobClient()
             blobClients[currentFile] = appendBlobClient
             appendBlobClient.create(true)
 
             val bufferedOutputStream =
-                    BufferedOutputStream(appendBlobClient.blobOutputStream, Math.toIntExact(GlobalDataSizeConstants.MAX_FILE_SIZE))
+                BufferedOutputStream(
+                    appendBlobClient.blobOutputStream,
+                    Math.toIntExact(GlobalDataSizeConstants.MAX_FILE_SIZE)
+                )
             val writer = PrintWriter(bufferedOutputStream, true, StandardCharsets.UTF_8)
             try {
                 csvPrinters[currentFile] = CSVPrinter(writer, CSVFormat.DEFAULT)
@@ -87,7 +92,12 @@ abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
     }
 
     private fun prepareAzureStagingFile(): String {
-        return java.lang.String.join("/", stagingFolder, schemaName, filenameGenerator.stagingFilename)
+        return java.lang.String.join(
+            "/",
+            stagingFolder,
+            schemaName,
+            filenameGenerator.stagingFilename
+        )
     }
 
     @Throws(Exception::class)
@@ -107,22 +117,48 @@ abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
 
     @Throws(Exception::class)
     override fun createTemporaryTable() {
-        LOGGER.info("Preparing tmp table in destination for stream: {}, schema: {}, tmp table name: {}.", streamName, schemaName, tmpTableName)
+        LOGGER.info(
+            "Preparing tmp table in destination for stream: {}, schema: {}, tmp table name: {}.",
+            streamName,
+            schemaName,
+            tmpTableName
+        )
         sqlOperations.createTableIfNotExists(db, schemaName, tmpTableName)
     }
 
     @Throws(Exception::class)
     override fun copyStagingFileToTemporaryTable() {
-        LOGGER.info("Starting copy to tmp table: {} in destination for stream: {}, schema: {}.", tmpTableName, streamName, schemaName)
+        LOGGER.info(
+            "Starting copy to tmp table: {} in destination for stream: {}, schema: {}.",
+            tmpTableName,
+            streamName,
+            schemaName
+        )
         for (azureStagingFile in azureStagingFiles) {
-            copyAzureBlobCsvFileIntoTable(db, getFullAzurePath(azureStagingFile), schemaName, tmpTableName, azureBlobConfig)
+            copyAzureBlobCsvFileIntoTable(
+                db,
+                getFullAzurePath(azureStagingFile),
+                schemaName,
+                tmpTableName,
+                azureBlobConfig
+            )
         }
-        LOGGER.info("Copy to tmp table {} in destination for stream {} complete.", tmpTableName, streamName)
+        LOGGER.info(
+            "Copy to tmp table {} in destination for stream {} complete.",
+            tmpTableName,
+            streamName
+        )
     }
 
     private fun getFullAzurePath(azureStagingFile: String?): String {
-        return ("azure://" + azureBlobConfig.accountName + "." + azureBlobConfig.endpointDomainName
-                + "/" + azureBlobConfig.containerName + "/" + azureStagingFile)
+        return ("azure://" +
+            azureBlobConfig.accountName +
+            "." +
+            azureBlobConfig.endpointDomainName +
+            "/" +
+            azureBlobConfig.containerName +
+            "/" +
+            azureStagingFile)
     }
 
     @Throws(Exception::class)
@@ -137,11 +173,20 @@ abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
 
     @Throws(Exception::class)
     override fun generateMergeStatement(destTableName: String?): String? {
-        LOGGER.info("Preparing to merge tmp table {} to dest table: {}, schema: {}, in destination.", tmpTableName, destTableName, schemaName)
+        LOGGER.info(
+            "Preparing to merge tmp table {} to dest table: {}, schema: {}, in destination.",
+            tmpTableName,
+            destTableName,
+            schemaName
+        )
         val queries = StringBuilder()
         if (destSyncMode == DestinationSyncMode.OVERWRITE) {
             queries.append(sqlOperations.truncateTableQuery(db, schemaName, destTableName))
-            LOGGER.info("Destination OVERWRITE mode detected. Dest table: {}, schema: {}, truncated.", destTableName, schemaName)
+            LOGGER.info(
+                "Destination OVERWRITE mode detected. Dest table: {}, schema: {}, truncated.",
+                destTableName,
+                schemaName
+            )
         }
         queries.append(sqlOperations.insertTableQuery(db, schemaName, tmpTableName, destTableName))
         return queries.toString()
@@ -175,18 +220,22 @@ abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
     }
 
     @Throws(SQLException::class)
-    abstract fun copyAzureBlobCsvFileIntoTable(database: JdbcDatabase?,
-                                               snowflakeAzureExternalStageName: String?,
-                                               schema: String?,
-                                               tableName: String?,
-                                               config: AzureBlobStorageConfig?)
+    abstract fun copyAzureBlobCsvFileIntoTable(
+        database: JdbcDatabase?,
+        snowflakeAzureExternalStageName: String?,
+        schema: String?,
+        tableName: String?,
+        config: AzureBlobStorageConfig?
+    )
 
     companion object {
-        private val LOGGER: Logger = LoggerFactory.getLogger(AzureBlobStorageStreamCopier::class.java)
+        private val LOGGER: Logger =
+            LoggerFactory.getLogger(AzureBlobStorageStreamCopier::class.java)
         fun attemptAzureBlobWriteAndDelete(config: AzureBlobStorageConfig) {
             var appendBlobClient: AppendBlobClient? = null
             try {
-                appendBlobClient = SpecializedBlobClientBuilder()
+                appendBlobClient =
+                    SpecializedBlobClientBuilder()
                         .endpoint(config.endpointUrl)
                         .sasToken(config.sasToken)
                         .containerName(config.containerName)
@@ -205,21 +254,34 @@ abstract class AzureBlobStorageStreamCopier(protected val stagingFolder: String,
         }
 
         private fun listCreatedBlob(containerClient: BlobContainerClient) {
-            containerClient.listBlobs().forEach(Consumer { blobItem: BlobItem -> LOGGER.info("Blob name: " + blobItem.name + "Snapshot: " + blobItem.snapshot) })
+            containerClient
+                .listBlobs()
+                .forEach(
+                    Consumer { blobItem: BlobItem ->
+                        LOGGER.info(
+                            "Blob name: " + blobItem.name + "Snapshot: " + blobItem.snapshot
+                        )
+                    }
+                )
         }
 
         private fun writeTestDataIntoBlob(appendBlobClient: AppendBlobClient?) {
             val test = "test_data"
             LOGGER.info("Writing test data to Azure Blob storage: $test")
-            val dataStream: InputStream = ByteArrayInputStream(test.toByteArray(StandardCharsets.UTF_8))
+            val dataStream: InputStream =
+                ByteArrayInputStream(test.toByteArray(StandardCharsets.UTF_8))
 
-            val blobCommittedBlockCount = appendBlobClient!!.appendBlock(dataStream, test.length.toLong())
+            val blobCommittedBlockCount =
+                appendBlobClient!!
+                    .appendBlock(dataStream, test.length.toLong())
                     .blobCommittedBlockCount
 
             LOGGER.info("blobCommittedBlockCount: $blobCommittedBlockCount")
         }
 
-        private fun getBlobContainerClient(appendBlobClient: AppendBlobClient?): BlobContainerClient {
+        private fun getBlobContainerClient(
+            appendBlobClient: AppendBlobClient?
+        ): BlobContainerClient {
             val containerClient = appendBlobClient!!.containerClient
             if (!containerClient.exists()) {
                 containerClient.create()
diff --git a/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageStreamCopierFactory.kt b/airbyte-cdk/java/airbyte-cdk/azure-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/copy/azure/AzureBlobStorageStreamCopierFactory.kt
@@ -13,39 +13,54 @@ import io.airbyte.cdk.integrations.destination.jdbc.copy.StreamCopierFactory.Com
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream
 import io.airbyte.protocol.models.v0.DestinationSyncMode
 
-abstract class AzureBlobStorageStreamCopierFactory : StreamCopierFactory<AzureBlobStorageConfig?> {
-    override fun create(configuredSchema: String?,
-                        azureBlobConfig: AzureBlobStorageConfig,
-                        stagingFolder: String?,
-                        configuredStream: ConfiguredAirbyteStream?,
-                        nameTransformer: StandardNameTransformer?,
-                        db: JdbcDatabase?,
-                        sqlOperations: SqlOperations?): StreamCopier? {
+abstract class AzureBlobStorageStreamCopierFactory : StreamCopierFactory<AzureBlobStorageConfig> {
+    override fun create(
+        configuredSchema: String?,
+        azureBlobConfig: AzureBlobStorageConfig,
+        stagingFolder: String?,
+        configuredStream: ConfiguredAirbyteStream?,
+        nameTransformer: StandardNameTransformer?,
+        db: JdbcDatabase?,
+        sqlOperations: SqlOperations?
+    ): StreamCopier? {
         try {
             val stream = configuredStream!!.stream
             val syncMode = configuredStream.destinationSyncMode
             val schema = getSchema(stream.namespace, configuredSchema!!, nameTransformer!!)
             val streamName = stream.name
 
-            val specializedBlobClientBuilder = SpecializedBlobClientBuilder()
+            val specializedBlobClientBuilder =
+                SpecializedBlobClientBuilder()
                     .endpoint(azureBlobConfig.endpointUrl)
                     .sasToken(azureBlobConfig.sasToken)
                     .containerName(azureBlobConfig.containerName)
 
-            return create(stagingFolder, syncMode, schema, streamName, specializedBlobClientBuilder, db, azureBlobConfig, nameTransformer, sqlOperations)
+            return create(
+                stagingFolder,
+                syncMode,
+                schema,
+                streamName,
+                specializedBlobClientBuilder,
+                db,
+                azureBlobConfig,
+                nameTransformer,
+                sqlOperations
+            )
         } catch (e: Exception) {
             throw RuntimeException(e)
         }
     }
 
     @Throws(Exception::class)
-    abstract fun create(stagingFolder: String?,
-                        syncMode: DestinationSyncMode?,
-                        schema: String?,
-                        streamName: String?,
-                        specializedBlobClientBuilder: SpecializedBlobClientBuilder?,
-                        db: JdbcDatabase?,
-                        azureBlobConfig: AzureBlobStorageConfig?,
-                        nameTransformer: StandardNameTransformer?,
-                        sqlOperations: SqlOperations?): StreamCopier?
+    abstract fun create(
+        stagingFolder: String?,
+        syncMode: DestinationSyncMode?,
+        schema: String?,
+        streamName: String?,
+        specializedBlobClientBuilder: SpecializedBlobClientBuilder?,
+        db: JdbcDatabase?,
+        azureBlobConfig: AzureBlobStorageConfig?,
+        nameTransformer: StandardNameTransformer?,
+        sqlOperations: SqlOperations?
+    ): StreamCopier?
 }
diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
@@ -1 +1 @@
-version=0.28.6
+version=0.28.7
